Merge branch 'master' of github.com:mattermost/mattermost-server into top-dms-clean
Этот коммит содержится в:
@@ -200,7 +200,7 @@ func (a *App) TestSiteURL(siteURL string) *model.AppError {
|
||||
return model.NewAppError("testSiteURL", "app.admin.test_site_url.failure", nil, "", http.StatusBadRequest)
|
||||
}
|
||||
defer func() {
|
||||
_, _ = io.Copy(ioutil.Discard, res.Body)
|
||||
_, _ = io.Copy(io.Discard, res.Body)
|
||||
_ = res.Body.Close()
|
||||
}()
|
||||
|
||||
|
||||
@@ -202,7 +202,7 @@ type AppIface interface {
|
||||
// lock instead.
|
||||
GetPluginsEnvironment() *plugin.Environment
|
||||
// GetPostsByIds response bool value indicates, if the post is inaccessible due to cloud plan's limit.
|
||||
GetPostsByIds(postIDs []string) ([]*model.Post, bool, *model.AppError)
|
||||
GetPostsByIds(postIDs []string) ([]*model.Post, int64, *model.AppError)
|
||||
// GetPostsUsage returns the total posts count rounded down to the most
|
||||
// significant digit
|
||||
GetPostsUsage() (int64, *model.AppError)
|
||||
@@ -446,6 +446,7 @@ type AppIface interface {
|
||||
CheckIntegrity() <-chan model.IntegrityCheckResult
|
||||
CheckMandatoryS3Fields(settings *model.FileSettings) *model.AppError
|
||||
CheckPasswordAndAllCriteria(user *model.User, password string, mfaToken string) *model.AppError
|
||||
CheckPostReminders()
|
||||
CheckRolesExist(roleNames []string) *model.AppError
|
||||
CheckUserAllAuthenticationCriteria(user *model.User, mfaToken string) *model.AppError
|
||||
CheckUserMfa(user *model.User, token string) *model.AppError
|
||||
@@ -630,8 +631,8 @@ type AppIface interface {
|
||||
GetFile(fileID string) ([]byte, *model.AppError)
|
||||
GetFileInfo(fileID string) (*model.FileInfo, *model.AppError)
|
||||
GetFileInfos(page, perPage int, opt *model.GetFileInfosOptions) ([]*model.FileInfo, *model.AppError)
|
||||
GetFileInfosForPost(postID string, fromMaster bool) ([]*model.FileInfo, *model.AppError)
|
||||
GetFileInfosForPostWithMigration(postID string) ([]*model.FileInfo, *model.AppError)
|
||||
GetFileInfosForPost(postID string, fromMaster bool, includeDeleted bool) ([]*model.FileInfo, *model.AppError)
|
||||
GetFileInfosForPostWithMigration(postID string, includeDeleted bool) ([]*model.FileInfo, *model.AppError)
|
||||
GetFlaggedPosts(userID string, offset int, limit int) (*model.PostList, *model.AppError)
|
||||
GetFlaggedPostsForChannel(userID, channelID string, offset int, limit int) (*model.PostList, *model.AppError)
|
||||
GetFlaggedPostsForTeam(userID, teamID string, offset int, limit int) (*model.PostList, *model.AppError)
|
||||
@@ -1042,6 +1043,7 @@ type AppIface interface {
|
||||
SetPluginKey(pluginID string, key string, value []byte) *model.AppError
|
||||
SetPluginKeyWithExpiry(pluginID string, key string, value []byte, expireInSeconds int64) *model.AppError
|
||||
SetPluginKeyWithOptions(pluginID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError)
|
||||
SetPostReminder(postID, userID string, targetTime int64) *model.AppError
|
||||
SetProfileImage(userID string, imageData *multipart.FileHeader) *model.AppError
|
||||
SetProfileImageFromFile(userID string, file io.Reader) *model.AppError
|
||||
SetProfileImageFromMultiPartFile(userID string, file multipart.File) *model.AppError
|
||||
|
||||
@@ -86,6 +86,9 @@ type Channels struct {
|
||||
|
||||
dndTaskMut sync.Mutex
|
||||
dndTask *model.ScheduledTask
|
||||
|
||||
postReminderMut sync.Mutex
|
||||
postReminderTask *model.ScheduledTask
|
||||
}
|
||||
|
||||
func init() {
|
||||
|
||||
@@ -82,9 +82,9 @@ func (w *configWrapper) SaveConfig(newCfg *model.Config, sendConfigChangeCluster
|
||||
if w.srv.Metrics != nil {
|
||||
w.srv.Metrics.Register()
|
||||
}
|
||||
w.srv.SetupMetricsServer()
|
||||
w.srv.platformService.RestartMetrics() // TODO: remove when this moved to the platform service
|
||||
} else {
|
||||
w.srv.StopMetricsServer()
|
||||
w.srv.platformService.ShutdownMetrics() // TODO: remove when this moved to the platform service
|
||||
}
|
||||
|
||||
if w.srv.Cluster != nil {
|
||||
|
||||
@@ -51,7 +51,7 @@ func (s *Server) downloadFromURL(downloadURL string) ([]byte, error) {
|
||||
}
|
||||
|
||||
if !(resp.StatusCode >= 200 && resp.StatusCode < 300) {
|
||||
_, _ = io.Copy(ioutil.Discard, resp.Body)
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
return errors.Errorf("failed to fetch from %s", downloadURL)
|
||||
}
|
||||
|
||||
@@ -1327,6 +1327,11 @@ func (a *App) SearchFilesInTeamForUser(c *request.Context, terms string, userId
|
||||
}
|
||||
|
||||
func (a *App) ExtractContentFromFileInfo(fileInfo *model.FileInfo) error {
|
||||
// We don't process images.
|
||||
if fileInfo.IsImage() {
|
||||
return nil
|
||||
}
|
||||
|
||||
file, aerr := a.FileReader(fileInfo.Path)
|
||||
if aerr != nil {
|
||||
return errors.Wrap(aerr, "failed to open file for extract file content")
|
||||
|
||||
@@ -543,3 +543,13 @@ func TestSearchFilesInTeamForUser(t *testing.T) {
|
||||
es.AssertExpectations(t)
|
||||
})
|
||||
}
|
||||
|
||||
func TestExtractContentFromFileInfo(t *testing.T) {
|
||||
app := &App{}
|
||||
fi := &model.FileInfo{
|
||||
MimeType: "image/jpeg",
|
||||
}
|
||||
|
||||
// Test that we don't process images.
|
||||
require.NoError(t, app.ExtractContentFromFileInfo(fi))
|
||||
}
|
||||
|
||||
@@ -1224,7 +1224,7 @@ func (a *App) importAttachment(c *request.Context, data *AttachmentImportData, p
|
||||
|
||||
// Go over existing files in the post and see if there already exists a file with the same name, size and hash. If so - skip it
|
||||
if post.Id != "" {
|
||||
oldFiles, err := a.GetFileInfosForPost(post.Id, true)
|
||||
oldFiles, err := a.GetFileInfosForPost(post.Id, true, false)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("BulkImport", "app.import.attachment.file_upload.error", map[string]any{"FilePath": *data.Path}, "", http.StatusBadRequest)
|
||||
}
|
||||
|
||||
@@ -331,6 +331,11 @@ func (a *App) AttachSessionCookies(c *request.Context, w http.ResponseWriter, r
|
||||
http.SetCookie(w, sessionCookie)
|
||||
http.SetCookie(w, userCookie)
|
||||
http.SetCookie(w, csrfCookie)
|
||||
|
||||
// For context see: https://mattermost.atlassian.net/browse/MM-39583
|
||||
if a.Channels().License() != nil && *a.Channels().License().Features.Cloud {
|
||||
a.AttachCloudSessionCookie(c, w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func GetProtocol(r *http.Request) string {
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"runtime"
|
||||
"strings"
|
||||
@@ -469,7 +468,7 @@ func (a *App) SendAckToPushProxy(ack *model.PushNotificationAck) error {
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
// Reading the body to completion.
|
||||
_, err = io.Copy(ioutil.Discard, resp.Body)
|
||||
_, err = io.Copy(io.Discard, resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -1310,6 +1310,21 @@ func (a *OpenTracingAppLayer) CheckPasswordAndAllCriteria(user *model.User, pass
|
||||
return resultVar0
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) CheckPostReminders() {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CheckPostReminders")
|
||||
|
||||
a.ctx = newCtx
|
||||
a.app.Srv().Store.SetContext(newCtx)
|
||||
defer func() {
|
||||
a.app.Srv().Store.SetContext(origCtx)
|
||||
a.ctx = origCtx
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
a.app.CheckPostReminders()
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) CheckProviderAttributes(user *model.User, patch *model.UserPatch) string {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CheckProviderAttributes")
|
||||
@@ -5970,7 +5985,7 @@ func (a *OpenTracingAppLayer) GetFileInfos(page int, perPage int, opt *model.Get
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetFileInfosForPost(postID string, fromMaster bool) ([]*model.FileInfo, *model.AppError) {
|
||||
func (a *OpenTracingAppLayer) GetFileInfosForPost(postID string, fromMaster bool, includeDeleted bool) ([]*model.FileInfo, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetFileInfosForPost")
|
||||
|
||||
@@ -5982,7 +5997,7 @@ func (a *OpenTracingAppLayer) GetFileInfosForPost(postID string, fromMaster bool
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
resultVar0, resultVar1 := a.app.GetFileInfosForPost(postID, fromMaster)
|
||||
resultVar0, resultVar1 := a.app.GetFileInfosForPost(postID, fromMaster, includeDeleted)
|
||||
|
||||
if resultVar1 != nil {
|
||||
span.LogFields(spanlog.Error(resultVar1))
|
||||
@@ -5992,7 +6007,7 @@ func (a *OpenTracingAppLayer) GetFileInfosForPost(postID string, fromMaster bool
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetFileInfosForPostWithMigration(postID string) ([]*model.FileInfo, *model.AppError) {
|
||||
func (a *OpenTracingAppLayer) GetFileInfosForPostWithMigration(postID string, includeDeleted bool) ([]*model.FileInfo, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetFileInfosForPostWithMigration")
|
||||
|
||||
@@ -6004,7 +6019,7 @@ func (a *OpenTracingAppLayer) GetFileInfosForPostWithMigration(postID string) ([
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
resultVar0, resultVar1 := a.app.GetFileInfosForPostWithMigration(postID)
|
||||
resultVar0, resultVar1 := a.app.GetFileInfosForPostWithMigration(postID, includeDeleted)
|
||||
|
||||
if resultVar1 != nil {
|
||||
span.LogFields(spanlog.Error(resultVar1))
|
||||
@@ -7881,7 +7896,7 @@ func (a *OpenTracingAppLayer) GetPostsBeforePost(options model.GetPostsOptions)
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetPostsByIds(postIDs []string) ([]*model.Post, bool, *model.AppError) {
|
||||
func (a *OpenTracingAppLayer) GetPostsByIds(postIDs []string) ([]*model.Post, int64, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetPostsByIds")
|
||||
|
||||
@@ -15614,6 +15629,28 @@ func (a *OpenTracingAppLayer) SetPluginKeyWithOptions(pluginID string, key strin
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) SetPostReminder(postID string, userID string, targetTime int64) *model.AppError {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SetPostReminder")
|
||||
|
||||
a.ctx = newCtx
|
||||
a.app.Srv().Store.SetContext(newCtx)
|
||||
defer func() {
|
||||
a.app.Srv().Store.SetContext(origCtx)
|
||||
a.ctx = origCtx
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
resultVar0 := a.app.SetPostReminder(postID, userID, targetTime)
|
||||
|
||||
if resultVar0 != nil {
|
||||
span.LogFields(spanlog.Error(resultVar0))
|
||||
ext.Error.Set(span, true)
|
||||
}
|
||||
|
||||
return resultVar0
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) SetProfileImage(userID string, imageData *multipart.FileHeader) *model.AppError {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SetProfileImage")
|
||||
|
||||
@@ -3,14 +3,28 @@
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/config"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
)
|
||||
|
||||
// ServiceConfig is used to initialize the PlatformService.
|
||||
// The mandatory fields will be checked during the initialization of the service.
|
||||
type ServiceConfig struct {
|
||||
// Mandatory fields
|
||||
ConfigStore *config.Store
|
||||
StartMetrics bool // TODO: find an elegant way to start/stop metrics server by default
|
||||
// Optional fields
|
||||
Metrics einterfaces.MetricsInterface
|
||||
Cluster einterfaces.ClusterInterface
|
||||
}
|
||||
|
||||
func (c *ServiceConfig) validate() error {
|
||||
// Mandatory fields need to be checked here
|
||||
if c.ConfigStore == nil {
|
||||
return errors.New("ConfigStore is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
162
app/platform/metrics.go
Обычный файл
162
app/platform/metrics.go
Обычный файл
@@ -0,0 +1,162 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/pprof"
|
||||
"runtime"
|
||||
"sync"
|
||||
"text/template"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
"github.com/mattermost/mattermost-server/v6/model"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
const TimeToWaitForConnectionsToCloseOnServerShutdown = time.Second
|
||||
|
||||
type platformMetrics struct {
|
||||
server *http.Server
|
||||
router *mux.Router
|
||||
lock sync.Mutex
|
||||
|
||||
metricsImpl einterfaces.MetricsInterface
|
||||
|
||||
cfgFn func() *model.Config
|
||||
}
|
||||
|
||||
func newPlatformMetrics(metricsImpl einterfaces.MetricsInterface, cfgFn func() *model.Config) *platformMetrics {
|
||||
if !*cfgFn().MetricsSettings.Enable {
|
||||
return nil
|
||||
}
|
||||
|
||||
pm := &platformMetrics{
|
||||
cfgFn: cfgFn,
|
||||
}
|
||||
|
||||
pm.stopMetricsServer()
|
||||
|
||||
if err := pm.initMetricsRouter(); err != nil {
|
||||
mlog.Error("Error initiating metrics router.", mlog.Err(err))
|
||||
}
|
||||
|
||||
if metricsImpl != nil {
|
||||
metricsImpl.Register()
|
||||
}
|
||||
|
||||
pm.startMetricsServer()
|
||||
|
||||
return pm
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) stopMetricsServer() {
|
||||
pm.lock.Lock()
|
||||
defer pm.lock.Unlock()
|
||||
|
||||
if pm.server != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), TimeToWaitForConnectionsToCloseOnServerShutdown)
|
||||
defer cancel()
|
||||
|
||||
pm.server.Shutdown(ctx)
|
||||
mlog.Info("Metrics and profiling server is stopping")
|
||||
}
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) startMetricsServer() {
|
||||
var notify chan struct{}
|
||||
pm.lock.Lock()
|
||||
defer func() {
|
||||
if notify != nil {
|
||||
<-notify
|
||||
}
|
||||
pm.lock.Unlock()
|
||||
}()
|
||||
|
||||
l, err := net.Listen("tcp", *pm.cfgFn().MetricsSettings.ListenAddress)
|
||||
if err != nil {
|
||||
mlog.Error(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
notify = make(chan struct{})
|
||||
pm.server = &http.Server{
|
||||
Handler: handlers.RecoveryHandler(handlers.PrintRecoveryStack(true))(pm.router),
|
||||
ReadTimeout: time.Duration(*pm.cfgFn().ServiceSettings.ReadTimeout) * time.Second,
|
||||
WriteTimeout: time.Duration(*pm.cfgFn().ServiceSettings.WriteTimeout) * time.Second,
|
||||
}
|
||||
|
||||
go func() {
|
||||
close(notify)
|
||||
if err := pm.server.Serve(l); err != nil && err != http.ErrServerClosed {
|
||||
mlog.Critical(err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
mlog.Info("Metrics and profiling server is started", mlog.String("address", l.Addr().String()))
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) initMetricsRouter() error {
|
||||
pm.router = mux.NewRouter()
|
||||
runtime.SetBlockProfileRate(*pm.cfgFn().MetricsSettings.BlockProfileRate)
|
||||
|
||||
metricsPage := `
|
||||
<html>
|
||||
<body>{{if .}}
|
||||
<div><a href="/metrics">Metrics</a></div>{{end}}
|
||||
<div><a href="/debug/pprof/">Profiling Root</a></div>
|
||||
<div><a href="/debug/pprof/cmdline">Profiling Command Line</a></div>
|
||||
<div><a href="/debug/pprof/symbol">Profiling Symbols</a></div>
|
||||
<div><a href="/debug/pprof/goroutine">Profiling Goroutines</a></div>
|
||||
<div><a href="/debug/pprof/heap">Profiling Heap</a></div>
|
||||
<div><a href="/debug/pprof/threadcreate">Profiling Threads</a></div>
|
||||
<div><a href="/debug/pprof/block">Profiling Blocking</a></div>
|
||||
<div><a href="/debug/pprof/trace">Profiling Execution Trace</a></div>
|
||||
<div><a href="/debug/pprof/profile">Profiling CPU</a></div>
|
||||
</body>
|
||||
</html>
|
||||
`
|
||||
metricsPageTmpl, err := template.New("page").Parse(metricsPage)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create template")
|
||||
}
|
||||
|
||||
rootHandler := func(w http.ResponseWriter, r *http.Request) {
|
||||
metricsPageTmpl.Execute(w, pm.metricsImpl != nil)
|
||||
}
|
||||
|
||||
pm.router.HandleFunc("/", rootHandler)
|
||||
pm.router.StrictSlash(true)
|
||||
|
||||
pm.router.Handle("/debug", http.RedirectHandler("/", http.StatusMovedPermanently))
|
||||
pm.router.HandleFunc("/debug/pprof/", pprof.Index)
|
||||
pm.router.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||||
pm.router.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||
pm.router.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||
pm.router.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||
|
||||
// Manually add support for paths linked to by index page at /debug/pprof/
|
||||
pm.router.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine"))
|
||||
pm.router.Handle("/debug/pprof/heap", pprof.Handler("heap"))
|
||||
pm.router.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate"))
|
||||
pm.router.Handle("/debug/pprof/block", pprof.Handler("block"))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ps *PlatformService) HandleMetrics(route string, h http.Handler) {
|
||||
if ps.metrics.router != nil {
|
||||
ps.metrics.router.Handle(route, h)
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *PlatformService) RestartMetrics() {
|
||||
ps.metrics = newPlatformMetrics(ps.serviceConfig.Metrics, ps.serviceConfig.ConfigStore.Get)
|
||||
}
|
||||
@@ -3,17 +3,42 @@
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"github.com/mattermost/mattermost-server/v6/config"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
)
|
||||
|
||||
// PlatformService is the service for the platform related tasks. It is
|
||||
// responsible for non-entity related functionalities that are required
|
||||
// by a product such as database access, configuration access, licensing etc.
|
||||
type PlatformService struct {
|
||||
serviceConfig ServiceConfig
|
||||
configStore *config.Store
|
||||
|
||||
metrics *platformMetrics
|
||||
|
||||
cluster einterfaces.ClusterInterface
|
||||
}
|
||||
|
||||
// New creates a new PlatformService.
|
||||
func New(c ServiceConfig) (*PlatformService, error) {
|
||||
if err := c.validate(); err != nil {
|
||||
func New(sc ServiceConfig) (*PlatformService, error) {
|
||||
if err := sc.validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &PlatformService{}, nil
|
||||
ps := &PlatformService{
|
||||
serviceConfig: sc,
|
||||
configStore: sc.ConfigStore,
|
||||
cluster: sc.Cluster,
|
||||
}
|
||||
|
||||
ps.metrics = newPlatformMetrics(sc.Metrics, ps.configStore.Get)
|
||||
|
||||
return ps, nil
|
||||
}
|
||||
|
||||
func (ps *PlatformService) ShutdownMetrics() {
|
||||
if ps.metrics != nil {
|
||||
ps.metrics.stopMetricsServer()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -513,7 +513,7 @@ func (api *PluginAPI) SearchPostsInTeam(teamID string, paramsList []*model.Searc
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return postList.ToSlice(), nil
|
||||
return postList.ForPlugin().ToSlice(), nil
|
||||
}
|
||||
|
||||
func (api *PluginAPI) SearchPostsInTeamForUser(teamID string, userID string, searchParams model.SearchParameter) (*model.PostSearchResults, *model.AppError) {
|
||||
@@ -547,7 +547,11 @@ func (api *PluginAPI) SearchPostsInTeamForUser(teamID string, userID string, sea
|
||||
includeDeletedChannels = *searchParams.IncludeDeletedChannels
|
||||
}
|
||||
|
||||
return api.app.SearchPostsForUser(api.ctx, terms, userID, teamID, isOrSearch, includeDeletedChannels, timeZoneOffset, page, perPage, model.ModifierMessages)
|
||||
results, appErr := api.app.SearchPostsForUser(api.ctx, terms, userID, teamID, isOrSearch, includeDeletedChannels, timeZoneOffset, page, perPage, model.ModifierMessages)
|
||||
if results != nil {
|
||||
results = results.ForPlugin()
|
||||
}
|
||||
return results, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) AddChannelMember(channelID, userID string) (*model.ChannelMember, *model.AppError) {
|
||||
@@ -627,7 +631,11 @@ func (api *PluginAPI) GetGroupsForUser(userID string) ([]*model.Group, *model.Ap
|
||||
}
|
||||
|
||||
func (api *PluginAPI) CreatePost(post *model.Post) (*model.Post, *model.AppError) {
|
||||
return api.app.CreatePostMissingChannel(api.ctx, post, true)
|
||||
post, appErr := api.app.CreatePostMissingChannel(api.ctx, post, true)
|
||||
if post != nil {
|
||||
post = post.ForPlugin()
|
||||
}
|
||||
return post, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) AddReaction(reaction *model.Reaction) (*model.Reaction, *model.AppError) {
|
||||
@@ -643,11 +651,11 @@ func (api *PluginAPI) GetReactions(postID string) ([]*model.Reaction, *model.App
|
||||
}
|
||||
|
||||
func (api *PluginAPI) SendEphemeralPost(userID string, post *model.Post) *model.Post {
|
||||
return api.app.SendEphemeralPost(api.ctx, userID, post)
|
||||
return api.app.SendEphemeralPost(api.ctx, userID, post).ForPlugin()
|
||||
}
|
||||
|
||||
func (api *PluginAPI) UpdateEphemeralPost(userID string, post *model.Post) *model.Post {
|
||||
return api.app.UpdateEphemeralPost(api.ctx, userID, post)
|
||||
return api.app.UpdateEphemeralPost(api.ctx, userID, post).ForPlugin()
|
||||
}
|
||||
|
||||
func (api *PluginAPI) DeleteEphemeralPost(userID, postID string) {
|
||||
@@ -660,31 +668,59 @@ func (api *PluginAPI) DeletePost(postID string) *model.AppError {
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPostThread(postID string) (*model.PostList, *model.AppError) {
|
||||
return api.app.GetPostThread(postID, model.GetPostsOptions{}, "")
|
||||
list, appErr := api.app.GetPostThread(postID, model.GetPostsOptions{}, "")
|
||||
if list != nil {
|
||||
list = list.ForPlugin()
|
||||
}
|
||||
return list, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPost(postID string) (*model.Post, *model.AppError) {
|
||||
return api.app.GetSinglePost(postID, false)
|
||||
post, appErr := api.app.GetSinglePost(postID, false)
|
||||
if post != nil {
|
||||
post = post.ForPlugin()
|
||||
}
|
||||
return post, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPostsSince(channelID string, time int64) (*model.PostList, *model.AppError) {
|
||||
return api.app.GetPostsSince(model.GetPostsSinceOptions{ChannelId: channelID, Time: time})
|
||||
list, appErr := api.app.GetPostsSince(model.GetPostsSinceOptions{ChannelId: channelID, Time: time})
|
||||
if list != nil {
|
||||
list = list.ForPlugin()
|
||||
}
|
||||
return list, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPostsAfter(channelID, postID string, page, perPage int) (*model.PostList, *model.AppError) {
|
||||
return api.app.GetPostsAfterPost(model.GetPostsOptions{ChannelId: channelID, PostId: postID, Page: page, PerPage: perPage})
|
||||
list, appErr := api.app.GetPostsAfterPost(model.GetPostsOptions{ChannelId: channelID, PostId: postID, Page: page, PerPage: perPage})
|
||||
if list != nil {
|
||||
list = list.ForPlugin()
|
||||
}
|
||||
return list, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPostsBefore(channelID, postID string, page, perPage int) (*model.PostList, *model.AppError) {
|
||||
return api.app.GetPostsBeforePost(model.GetPostsOptions{ChannelId: channelID, PostId: postID, Page: page, PerPage: perPage})
|
||||
list, appErr := api.app.GetPostsBeforePost(model.GetPostsOptions{ChannelId: channelID, PostId: postID, Page: page, PerPage: perPage})
|
||||
if list != nil {
|
||||
list = list.ForPlugin()
|
||||
}
|
||||
return list, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetPostsForChannel(channelID string, page, perPage int) (*model.PostList, *model.AppError) {
|
||||
return api.app.GetPostsPage(model.GetPostsOptions{ChannelId: channelID, Page: page, PerPage: perPage})
|
||||
list, appErr := api.app.GetPostsPage(model.GetPostsOptions{ChannelId: channelID, Page: page, PerPage: perPage})
|
||||
if list != nil {
|
||||
list = list.ForPlugin()
|
||||
}
|
||||
return list, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) UpdatePost(post *model.Post) (*model.Post, *model.AppError) {
|
||||
return api.app.UpdatePost(api.ctx, post, false)
|
||||
post, appErr := api.app.UpdatePost(api.ctx, post, false)
|
||||
if post != nil {
|
||||
post = post.ForPlugin()
|
||||
}
|
||||
return post, appErr
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetProfileImage(userID string) ([]byte, *model.AppError) {
|
||||
|
||||
@@ -770,6 +770,44 @@ func TestPluginPanicLogs(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestPluginStatusActivateError(t *testing.T) {
|
||||
t.Run("should return error from OnActivate in plugin statuses", func(t *testing.T) {
|
||||
th := Setup(t).InitBasic()
|
||||
defer th.TearDown()
|
||||
|
||||
pluginSource := `
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/plugin"
|
||||
)
|
||||
|
||||
type MyPlugin struct {
|
||||
plugin.MattermostPlugin
|
||||
}
|
||||
|
||||
func (p *MyPlugin) OnActivate() error {
|
||||
return errors.New("sample error")
|
||||
}
|
||||
|
||||
func main() {
|
||||
plugin.ClientMain(&MyPlugin{})
|
||||
}
|
||||
`
|
||||
|
||||
tearDown, _, _ := SetAppEnvironmentWithPlugins(t, []string{pluginSource}, th.App, th.NewPluginAPI)
|
||||
defer tearDown()
|
||||
|
||||
env := th.App.GetPluginsEnvironment()
|
||||
pluginStatus, err := env.Statuses()
|
||||
require.NoError(t, err)
|
||||
require.Len(t, pluginStatus, 1)
|
||||
require.Equal(t, "sample error", pluginStatus[0].Error)
|
||||
})
|
||||
}
|
||||
|
||||
func TestProcessPrepackagedPlugins(t *testing.T) {
|
||||
th := Setup(t)
|
||||
defer th.TearDown()
|
||||
|
||||
175
app/post.go
175
app/post.go
@@ -259,7 +259,7 @@ func (a *App) CreatePost(c request.CTX, post *model.Post, channel *model.Channel
|
||||
var rejectionError *model.AppError
|
||||
pluginContext := pluginContext(c)
|
||||
pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
|
||||
replacementPost, rejectionReason := hooks.MessageWillBePosted(pluginContext, post)
|
||||
replacementPost, rejectionReason := hooks.MessageWillBePosted(pluginContext, post.ForPlugin())
|
||||
if rejectionReason != "" {
|
||||
id := "Post rejected by plugin. " + rejectionReason
|
||||
if rejectionReason == plugin.DismissPostError {
|
||||
@@ -269,6 +269,7 @@ func (a *App) CreatePost(c request.CTX, post *model.Post, channel *model.Channel
|
||||
return false
|
||||
}
|
||||
if replacementPost != nil {
|
||||
// the original post's metadata (if there ever was any) is lost, and will be rebuilt.
|
||||
post = replacementPost
|
||||
}
|
||||
|
||||
@@ -309,21 +310,14 @@ func (a *App) CreatePost(c request.CTX, post *model.Post, channel *model.Channel
|
||||
// might be duplicating requests.
|
||||
a.Srv().seenPendingPostIdsCache.SetWithExpiry(post.PendingPostId, rpost.Id, PendingPostIDsCacheTTL)
|
||||
|
||||
// We make a copy of the post for the plugin hook to avoid a race condition.
|
||||
rPostCopy := rpost.Clone()
|
||||
|
||||
// FIXME: Removes PreviewPost from the post payload sent to the MessageHasBeenPosted hook so that plugins compiled with older versions of
|
||||
// Mattermost—without the gob registration of the PreviewPost struct—won't crash.
|
||||
if rPostCopy.Metadata != nil {
|
||||
rPostCopy.Metadata = rPostCopy.Metadata.Copy()
|
||||
}
|
||||
rPostCopy.RemovePreviewPost()
|
||||
|
||||
// We make a copy of the post for the plugin hook to avoid a race condition,
|
||||
// and to remove the non-GOB-encodable Metadata from it.
|
||||
if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
|
||||
pluginPost := rpost.ForPlugin()
|
||||
a.Srv().Go(func() {
|
||||
pluginContext := pluginContext(c)
|
||||
pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
|
||||
hooks.MessageHasBeenPosted(pluginContext, rPostCopy)
|
||||
hooks.MessageHasBeenPosted(pluginContext, pluginPost)
|
||||
return true
|
||||
}, plugin.MessageHasBeenPostedID)
|
||||
})
|
||||
@@ -650,12 +644,15 @@ func (a *App) UpdatePost(c *request.Context, post *model.Post, safeUpdate bool)
|
||||
var rejectionReason string
|
||||
pluginContext := pluginContext(c)
|
||||
pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
|
||||
newPost, rejectionReason = hooks.MessageWillBeUpdated(pluginContext, newPost, oldPost)
|
||||
newPost, rejectionReason = hooks.MessageWillBeUpdated(pluginContext, newPost.ForPlugin(), oldPost.ForPlugin())
|
||||
return post != nil
|
||||
}, plugin.MessageWillBeUpdatedID)
|
||||
if newPost == nil {
|
||||
return nil, model.NewAppError("UpdatePost", "Post rejected by plugin. "+rejectionReason, nil, "", http.StatusBadRequest)
|
||||
}
|
||||
// Restore the post metadata that was stripped by the plugin. Set it to
|
||||
// the last known good.
|
||||
newPost.Metadata = oldPost.Metadata
|
||||
}
|
||||
|
||||
rpost, nErr := a.Srv().Store.Post().Update(newPost, oldPost)
|
||||
@@ -670,10 +667,12 @@ func (a *App) UpdatePost(c *request.Context, post *model.Post, safeUpdate bool)
|
||||
}
|
||||
|
||||
if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
|
||||
pluginOldPost := oldPost.ForPlugin()
|
||||
pluginNewPost := newPost.ForPlugin()
|
||||
a.Srv().Go(func() {
|
||||
pluginContext := pluginContext(c)
|
||||
pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
|
||||
hooks.MessageHasBeenUpdated(pluginContext, newPost, oldPost)
|
||||
hooks.MessageHasBeenUpdated(pluginContext, pluginNewPost, pluginOldPost)
|
||||
return true
|
||||
}, plugin.MessageHasBeenUpdatedID)
|
||||
})
|
||||
@@ -871,11 +870,11 @@ func (a *App) GetSinglePost(postID string, includeDeleted bool) (*model.Post, *m
|
||||
}
|
||||
}
|
||||
|
||||
isInaccessible, appErr := a.isInaccessiblePost(post)
|
||||
firstInaccessiblePostTime, appErr := a.isInaccessiblePost(post)
|
||||
if appErr != nil {
|
||||
return nil, appErr
|
||||
}
|
||||
if isInaccessible {
|
||||
if firstInaccessiblePostTime != 0 {
|
||||
return nil, model.NewAppError("GetSinglePost", "app.post.cloud.get.app_error", nil, "", http.StatusForbidden)
|
||||
}
|
||||
|
||||
@@ -1270,6 +1269,8 @@ func (a *App) DeletePost(c request.CTX, postID, deleteByID string) (*model.Post,
|
||||
a.Srv().Go(func() {
|
||||
a.deletePostFiles(post.Id)
|
||||
})
|
||||
a.Srv().Store.FileInfo().InvalidateFileInfosForPostCache(postID, true)
|
||||
a.Srv().Store.FileInfo().InvalidateFileInfosForPostCache(postID, false)
|
||||
}
|
||||
a.Srv().Go(func() {
|
||||
a.deleteFlaggedPosts(post.Id)
|
||||
@@ -1542,16 +1543,16 @@ func (a *App) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, *m
|
||||
return searchParams, nil
|
||||
}
|
||||
|
||||
func (a *App) GetFileInfosForPostWithMigration(postID string) ([]*model.FileInfo, *model.AppError) {
|
||||
func (a *App) GetFileInfosForPostWithMigration(postID string, includeDeleted bool) ([]*model.FileInfo, *model.AppError) {
|
||||
|
||||
pchan := make(chan store.StoreResult, 1)
|
||||
go func() {
|
||||
post, err := a.Srv().Store.Post().GetSingle(postID, false)
|
||||
post, err := a.Srv().Store.Post().GetSingle(postID, includeDeleted)
|
||||
pchan <- store.StoreResult{Data: post, NErr: err}
|
||||
close(pchan)
|
||||
}()
|
||||
|
||||
infos, err := a.GetFileInfosForPost(postID, false)
|
||||
infos, err := a.GetFileInfosForPost(postID, false, includeDeleted)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1581,8 +1582,8 @@ func (a *App) GetFileInfosForPostWithMigration(postID string) ([]*model.FileInfo
|
||||
return infos, nil
|
||||
}
|
||||
|
||||
func (a *App) GetFileInfosForPost(postID string, fromMaster bool) ([]*model.FileInfo, *model.AppError) {
|
||||
fileInfos, err := a.Srv().Store.FileInfo().GetForPost(postID, fromMaster, false, true)
|
||||
func (a *App) GetFileInfosForPost(postID string, fromMaster bool, includeDeleted bool) ([]*model.FileInfo, *model.AppError) {
|
||||
fileInfos, err := a.Srv().Store.FileInfo().GetForPost(postID, fromMaster, includeDeleted, true)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("GetFileInfosForPost", "app.file_info.get_for_post.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
@@ -1880,24 +1881,24 @@ func (a *App) GetPostIfAuthorized(c request.CTX, postID string, session *model.S
|
||||
}
|
||||
|
||||
// GetPostsByIds response bool value indicates, if the post is inaccessible due to cloud plan's limit.
|
||||
func (a *App) GetPostsByIds(postIDs []string) ([]*model.Post, bool, *model.AppError) {
|
||||
func (a *App) GetPostsByIds(postIDs []string) ([]*model.Post, int64, *model.AppError) {
|
||||
posts, err := a.Srv().Store.Post().GetPostsByIds(postIDs)
|
||||
if err != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
switch {
|
||||
case errors.As(err, &nfErr):
|
||||
return nil, false, model.NewAppError("GetPostsByIds", "app.post.get.app_error", nil, nfErr.Error(), http.StatusNotFound)
|
||||
return nil, 0, model.NewAppError("GetPostsByIds", "app.post.get.app_error", nil, nfErr.Error(), http.StatusNotFound)
|
||||
default:
|
||||
return nil, false, model.NewAppError("GetPostsByIds", "app.post.get.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
return nil, 0, model.NewAppError("GetPostsByIds", "app.post.get.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
}
|
||||
|
||||
posts, hasInaccessiblePosts, appErr := a.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
posts, firstInaccessiblePostTime, appErr := a.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
if appErr != nil {
|
||||
return nil, false, appErr
|
||||
return nil, 0, appErr
|
||||
}
|
||||
|
||||
return posts, hasInaccessiblePosts, nil
|
||||
return posts, firstInaccessiblePostTime, nil
|
||||
}
|
||||
|
||||
func (a *App) GetTopThreadsForTeamSince(c request.CTX, teamID, userID string, opts *model.InsightsOpts) (*model.TopThreadList, *model.AppError) {
|
||||
@@ -1943,6 +1944,126 @@ func (a *App) GetTopDMsForUserSince(userID string, opts *model.InsightsOpts) (*m
|
||||
return topDMs, nil
|
||||
}
|
||||
|
||||
func (a *App) SetPostReminder(postID, userID string, targetTime int64) *model.AppError {
|
||||
// Store the reminder in the DB
|
||||
reminder := &model.PostReminder{
|
||||
PostId: postID,
|
||||
UserId: userID,
|
||||
TargetTime: targetTime,
|
||||
}
|
||||
err := a.Srv().Store.Post().SetPostReminder(reminder)
|
||||
if err != nil {
|
||||
return model.NewAppError("SetPostReminder", "app.post_reminder.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
metadata, err := a.Srv().Store.Post().GetPostReminderMetadata(postID)
|
||||
if err != nil {
|
||||
return model.NewAppError("SetPostReminder", "app.post_reminder.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
parsed := time.Unix(targetTime, 0).UTC().Format(time.RFC822)
|
||||
siteURL := *a.Config().ServiceSettings.SiteURL
|
||||
// Send an ack message.
|
||||
ephemeralPost := &model.Post{
|
||||
Type: model.PostTypeEphemeral,
|
||||
Id: model.NewId(),
|
||||
CreateAt: model.GetMillis(),
|
||||
UserId: userID,
|
||||
RootId: postID,
|
||||
ChannelId: metadata.ChannelId,
|
||||
// It's okay to keep this non-translated. This is just a fallback.
|
||||
// The webapp will parse the timestamp and show that in user's local timezone.
|
||||
Message: fmt.Sprintf("You will be reminded about %s/%s/pl/%s by @%s at %s", siteURL, metadata.TeamName, postID, metadata.Username, parsed),
|
||||
Props: model.StringInterface{
|
||||
"target_time": targetTime,
|
||||
"team_name": metadata.TeamName,
|
||||
"post_id": postID,
|
||||
"username": metadata.Username,
|
||||
"type": model.PostTypeReminder,
|
||||
},
|
||||
}
|
||||
|
||||
message := model.NewWebSocketEvent(model.WebsocketEventEphemeralMessage, "", ephemeralPost.ChannelId, userID, nil)
|
||||
ephemeralPost = a.PreparePostForClientWithEmbedsAndImages(request.EmptyContext(a.Log()), ephemeralPost, true, false)
|
||||
ephemeralPost = model.AddPostActionCookies(ephemeralPost, a.PostActionCookieSecret())
|
||||
|
||||
postJSON, jsonErr := ephemeralPost.ToJSON()
|
||||
if jsonErr != nil {
|
||||
mlog.Warn("Failed to encode post to JSON", mlog.Err(jsonErr))
|
||||
}
|
||||
message.Add("post", postJSON)
|
||||
a.Publish(message)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *App) CheckPostReminders() {
|
||||
systemBot, appErr := a.GetSystemBot()
|
||||
if appErr != nil {
|
||||
mlog.Error("Failed to get system bot", mlog.Err(appErr))
|
||||
return
|
||||
}
|
||||
|
||||
// This will return the reminders and also delete them from the DB.
|
||||
// In case, any of the next steps fail, those reminders would be lost.
|
||||
// Alternatively, if we delete those reminders _after_ it has been sent,
|
||||
// then in case of any temporary failure, they would get sent in the next batch.
|
||||
// MM-45595.
|
||||
reminders, err := a.Srv().Store.Post().GetPostReminders(time.Now().UTC().Unix())
|
||||
if err != nil {
|
||||
mlog.Error("Failed to get post reminders", mlog.Err(err))
|
||||
return
|
||||
}
|
||||
|
||||
// We group multiple reminders for a single user.
|
||||
groupedReminders := make(map[string][]string)
|
||||
for _, r := range reminders {
|
||||
if groupedReminders[r.UserId] == nil {
|
||||
groupedReminders[r.UserId] = []string{r.PostId}
|
||||
} else {
|
||||
groupedReminders[r.UserId] = append(groupedReminders[r.UserId], r.PostId)
|
||||
}
|
||||
}
|
||||
|
||||
siteURL := *a.Config().ServiceSettings.SiteURL
|
||||
for userID, postIDs := range groupedReminders {
|
||||
ch, appErr := a.GetOrCreateDirectChannel(request.EmptyContext(a.Log()), userID, systemBot.UserId)
|
||||
if appErr != nil {
|
||||
mlog.Error("Failed to get direct channel", mlog.Err(appErr))
|
||||
return
|
||||
}
|
||||
|
||||
for _, postID := range postIDs {
|
||||
metadata, err := a.Srv().Store.Post().GetPostReminderMetadata(postID)
|
||||
if err != nil {
|
||||
mlog.Error("Failed to get post reminder metadata", mlog.Err(err))
|
||||
continue
|
||||
}
|
||||
|
||||
T := i18n.GetUserTranslations(metadata.UserLocale)
|
||||
dm := &model.Post{
|
||||
ChannelId: ch.Id,
|
||||
Message: T("app.post_reminder_dm", model.StringInterface{
|
||||
"SiteURL": siteURL,
|
||||
"TeamName": metadata.TeamName,
|
||||
"PostId": postID,
|
||||
"Username": metadata.Username,
|
||||
}),
|
||||
Type: model.PostTypeDefault,
|
||||
UserId: systemBot.UserId,
|
||||
Props: model.StringInterface{
|
||||
"username": systemBot.Username,
|
||||
},
|
||||
}
|
||||
|
||||
if _, err := a.CreatePost(request.EmptyContext(a.Log()), dm, ch, false, true); err != nil {
|
||||
mlog.Error("Failed to post reminder message", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func includeEmbedsAndImages(a *App, c request.CTX, topThreadList *model.TopThreadList, userID string) (*model.TopThreadList, error) {
|
||||
for _, topThread := range topThreadList.Items {
|
||||
topThread.Post = a.PreparePostForClientWithEmbedsAndImages(c, topThread.Post, false, false)
|
||||
|
||||
@@ -27,6 +27,19 @@ func (b accessibleBounds) noAccessible() bool {
|
||||
return b.start == noAccessibleBounds.start && b.end == noAccessibleBounds.end
|
||||
}
|
||||
|
||||
// assumes checking was already performed that at least one post is inaccessible
|
||||
func (b accessibleBounds) getInaccessibleRange(listLength int) (int, int) {
|
||||
var start, end int
|
||||
if b.start == 0 {
|
||||
start = b.end + 1
|
||||
end = listLength - 1
|
||||
} else {
|
||||
start = 0
|
||||
end = b.start - 1
|
||||
}
|
||||
return start, end
|
||||
}
|
||||
|
||||
var noAccessibleBounds = accessibleBounds{start: -1, end: -1}
|
||||
var allAccessibleBounds = func(lenPosts int) accessibleBounds { return accessibleBounds{start: 0, end: lenPosts - 1} }
|
||||
|
||||
@@ -82,11 +95,13 @@ func linearFilterPostList(postList *model.PostList, earliestAccessibleTime int64
|
||||
|
||||
n := 0
|
||||
for i, postId := range order {
|
||||
if posts[postId].CreateAt >= earliestAccessibleTime {
|
||||
if createAt := posts[postId].CreateAt; createAt >= earliestAccessibleTime {
|
||||
order[n] = order[i]
|
||||
n++
|
||||
} else {
|
||||
postList.HasInaccessiblePosts = true
|
||||
if createAt > postList.FirstInaccessiblePostTime {
|
||||
postList.FirstInaccessiblePostTime = createAt
|
||||
}
|
||||
delete(posts, postId)
|
||||
}
|
||||
}
|
||||
@@ -96,8 +111,10 @@ func linearFilterPostList(postList *model.PostList, earliestAccessibleTime int64
|
||||
// for example GetPosts in the CollapsedThreads = false path, parents are not added
|
||||
// to Order
|
||||
for postId := range posts {
|
||||
if posts[postId].CreateAt < earliestAccessibleTime {
|
||||
postList.HasInaccessiblePosts = true
|
||||
if createAt := posts[postId].CreateAt; createAt < earliestAccessibleTime {
|
||||
if createAt > postList.FirstInaccessiblePostTime {
|
||||
postList.FirstInaccessiblePostTime = createAt
|
||||
}
|
||||
delete(posts, postId)
|
||||
}
|
||||
}
|
||||
@@ -106,18 +123,20 @@ func linearFilterPostList(postList *model.PostList, earliestAccessibleTime int64
|
||||
// linearFilterPostsSlice make no assumptions about ordering, go through posts one by one
|
||||
// this is the slower fallback that is still safe if we can not
|
||||
// assume posts are ordered by CreatedAt
|
||||
func linearFilterPostsSlice(posts []*model.Post, earliestAccessibleTime int64) ([]*model.Post, bool) {
|
||||
hasInaccessiblePosts := false
|
||||
func linearFilterPostsSlice(posts []*model.Post, earliestAccessibleTime int64) ([]*model.Post, int64) {
|
||||
var firstInaccessiblePostTime int64 = 0
|
||||
n := 0
|
||||
for i := range posts {
|
||||
if posts[i].CreateAt >= earliestAccessibleTime {
|
||||
if createAt := posts[i].CreateAt; createAt >= earliestAccessibleTime {
|
||||
posts[n] = posts[i]
|
||||
n++
|
||||
} else {
|
||||
hasInaccessiblePosts = true
|
||||
if createAt > firstInaccessiblePostTime {
|
||||
firstInaccessiblePostTime = createAt
|
||||
}
|
||||
}
|
||||
}
|
||||
return posts[:n], hasInaccessiblePosts
|
||||
return posts[:n], firstInaccessiblePostTime
|
||||
}
|
||||
|
||||
// filterInaccessiblePosts filters out the posts, past the cloud limit
|
||||
@@ -146,13 +165,18 @@ func (a *App) filterInaccessiblePosts(postList *model.PostList, options filterPo
|
||||
}
|
||||
if bounds.noAccessible() {
|
||||
if lenPosts > 0 {
|
||||
postList.HasInaccessiblePosts = true
|
||||
firstPostCreatedAt := postList.Posts[postList.Order[0]].CreateAt
|
||||
lastPostCreatedAt := postList.Posts[postList.Order[len(postList.Order)-1]].CreateAt
|
||||
postList.FirstInaccessiblePostTime = max(firstPostCreatedAt, lastPostCreatedAt)
|
||||
}
|
||||
postList.Posts = map[string]*model.Post{}
|
||||
postList.Order = []string{}
|
||||
return nil
|
||||
}
|
||||
postList.HasInaccessiblePosts = true
|
||||
startInaccessibleIndex, endInaccessibleIndex := bounds.getInaccessibleRange(len(postList.Order))
|
||||
startInaccessibleCreatedAt := postList.Posts[postList.Order[startInaccessibleIndex]].CreateAt
|
||||
endInaccessibleCreatedAt := postList.Posts[postList.Order[endInaccessibleIndex]].CreateAt
|
||||
postList.FirstInaccessiblePostTime = max(startInaccessibleCreatedAt, endInaccessibleCreatedAt)
|
||||
|
||||
posts := postList.Posts
|
||||
order := postList.Order
|
||||
@@ -183,9 +207,9 @@ func (a *App) filterInaccessiblePosts(postList *model.PostList, options filterPo
|
||||
}
|
||||
|
||||
// isInaccessiblePost indicates if the post is past the cloud plan's limit.
|
||||
func (a *App) isInaccessiblePost(post *model.Post) (bool, *model.AppError) {
|
||||
func (a *App) isInaccessiblePost(post *model.Post) (int64, *model.AppError) {
|
||||
if post == nil {
|
||||
return false, nil
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
pl := &model.PostList{
|
||||
@@ -193,22 +217,22 @@ func (a *App) isInaccessiblePost(post *model.Post) (bool, *model.AppError) {
|
||||
Posts: map[string]*model.Post{post.Id: post},
|
||||
}
|
||||
|
||||
return pl.HasInaccessiblePosts, a.filterInaccessiblePosts(pl, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
return pl.FirstInaccessiblePostTime, a.filterInaccessiblePosts(pl, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
}
|
||||
|
||||
// getFilteredAccessiblePosts returns accessible posts filtered as per the cloud plan's limit and also indicates if there were any inaccessible posts
|
||||
func (a *App) getFilteredAccessiblePosts(posts []*model.Post, options filterPostOptions) ([]*model.Post, bool, *model.AppError) {
|
||||
func (a *App) getFilteredAccessiblePosts(posts []*model.Post, options filterPostOptions) ([]*model.Post, int64, *model.AppError) {
|
||||
if len(posts) == 0 {
|
||||
return posts, false, nil
|
||||
return posts, 0, nil
|
||||
}
|
||||
|
||||
filteredPosts := []*model.Post{}
|
||||
lastAccessiblePostTime, appErr := a.GetLastAccessiblePostTime()
|
||||
if appErr != nil {
|
||||
return filteredPosts, false, model.NewAppError("getFilteredAccessiblePosts", "app.last_accessible_post.app_error", nil, appErr.Error(), http.StatusInternalServerError)
|
||||
return filteredPosts, 0, model.NewAppError("getFilteredAccessiblePosts", "app.last_accessible_post.app_error", nil, appErr.Error(), http.StatusInternalServerError)
|
||||
} else if lastAccessiblePostTime == 0 {
|
||||
// No need to filter, all posts are accessible
|
||||
return posts, false, nil
|
||||
return posts, 0, nil
|
||||
}
|
||||
|
||||
if options.assumeSortedCreatedAt {
|
||||
@@ -216,16 +240,26 @@ func (a *App) getFilteredAccessiblePosts(posts []*model.Post, options filterPost
|
||||
getCreateAt := func(i int) int64 { return posts[i].CreateAt }
|
||||
bounds := getTimeSortedPostAccessibleBounds(lastAccessiblePostTime, lenPosts, getCreateAt)
|
||||
if bounds.allAccessible(lenPosts) {
|
||||
return posts, false, nil
|
||||
return posts, 0, nil
|
||||
}
|
||||
if bounds.noAccessible() {
|
||||
return filteredPosts, lenPosts > 0, nil
|
||||
var firstInaccessiblePostTime int64 = 0
|
||||
if lenPosts > 0 {
|
||||
firstPostCreatedAt := posts[0].CreateAt
|
||||
lastPostCreatedAt := posts[len(posts)-1].CreateAt
|
||||
firstInaccessiblePostTime = max(firstPostCreatedAt, lastPostCreatedAt)
|
||||
}
|
||||
return filteredPosts, firstInaccessiblePostTime, nil
|
||||
}
|
||||
|
||||
startInaccessibleIndex, endInaccessibleIndex := bounds.getInaccessibleRange(len(posts))
|
||||
firstPostCreatedAt := posts[startInaccessibleIndex].CreateAt
|
||||
lastPostCreatedAt := posts[endInaccessibleIndex].CreateAt
|
||||
firstInaccessiblePostTime := max(firstPostCreatedAt, lastPostCreatedAt)
|
||||
filteredPosts = posts[bounds.start : bounds.end+1]
|
||||
return filteredPosts, true, nil
|
||||
return filteredPosts, firstInaccessiblePostTime, nil
|
||||
}
|
||||
|
||||
filteredPosts, hasInaccessiblePosts := linearFilterPostsSlice(posts, lastAccessiblePostTime)
|
||||
return filteredPosts, hasInaccessiblePosts, nil
|
||||
filteredPosts, firstInaccessiblePostTime := linearFilterPostsSlice(posts, lastAccessiblePostTime)
|
||||
return filteredPosts, firstInaccessiblePostTime, nil
|
||||
}
|
||||
|
||||
@@ -231,6 +231,7 @@ func TestFilterInaccessiblePosts(t *testing.T) {
|
||||
"post_d",
|
||||
"post_e",
|
||||
}, postList.Order)
|
||||
assert.Equal(t, int64(1), postList.FirstInaccessiblePostTime)
|
||||
})
|
||||
|
||||
t.Run("descending order returns correct posts", func(t *testing.T) {
|
||||
@@ -259,6 +260,8 @@ func TestFilterInaccessiblePosts(t *testing.T) {
|
||||
"post_d",
|
||||
"post_c",
|
||||
}, postList.Order)
|
||||
|
||||
assert.Equal(t, int64(1), postList.FirstInaccessiblePostTime)
|
||||
})
|
||||
|
||||
t.Run("handles mixed create at ordering correctly if correct options given", func(t *testing.T) {
|
||||
@@ -332,18 +335,20 @@ func TestGetFilteredAccessiblePosts(t *testing.T) {
|
||||
|
||||
t.Run("ascending order returns correct posts", func(t *testing.T) {
|
||||
posts := []*model.Post{postFromCreateAt(0), postFromCreateAt(1), postFromCreateAt(2), postFromCreateAt(3), postFromCreateAt(4)}
|
||||
filteredPosts, _, appErr := th.App.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
filteredPosts, firstInaccessiblePostTime, appErr := th.App.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
|
||||
assert.Nil(t, appErr)
|
||||
assert.Equal(t, []*model.Post{postFromCreateAt(2), postFromCreateAt(3), postFromCreateAt(4)}, filteredPosts)
|
||||
assert.Equal(t, int64(1), firstInaccessiblePostTime)
|
||||
})
|
||||
|
||||
t.Run("descending order returns correct posts", func(t *testing.T) {
|
||||
posts := []*model.Post{postFromCreateAt(4), postFromCreateAt(3), postFromCreateAt(2), postFromCreateAt(1), postFromCreateAt(0)}
|
||||
filteredPosts, _, appErr := th.App.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
filteredPosts, firstInaccessiblePostTime, appErr := th.App.getFilteredAccessiblePosts(posts, filterPostOptions{assumeSortedCreatedAt: true})
|
||||
|
||||
assert.Nil(t, appErr)
|
||||
assert.Equal(t, []*model.Post{postFromCreateAt(4), postFromCreateAt(3), postFromCreateAt(2)}, filteredPosts)
|
||||
assert.Equal(t, int64(1), firstInaccessiblePostTime)
|
||||
})
|
||||
|
||||
t.Run("handles mixed create at ordering correctly if correct options given", func(t *testing.T) {
|
||||
@@ -366,12 +371,40 @@ func TestIsInaccessiblePost(t *testing.T) {
|
||||
defer th.TearDown()
|
||||
|
||||
post := &model.Post{CreateAt: 3}
|
||||
r, appErr := th.App.isInaccessiblePost(post)
|
||||
firstInaccessiblePostTime, appErr := th.App.isInaccessiblePost(post)
|
||||
assert.Nil(t, appErr)
|
||||
assert.Equal(t, false, r)
|
||||
assert.Equal(t, int64(0), firstInaccessiblePostTime)
|
||||
|
||||
post = &model.Post{CreateAt: 1}
|
||||
r, appErr = th.App.isInaccessiblePost(post)
|
||||
firstInaccessiblePostTime, appErr = th.App.isInaccessiblePost(post)
|
||||
assert.Nil(t, appErr)
|
||||
assert.Equal(t, true, r)
|
||||
assert.Equal(t, int64(1), firstInaccessiblePostTime)
|
||||
}
|
||||
|
||||
func Test_getInaccessibleRange(t *testing.T) {
|
||||
type test struct {
|
||||
label string
|
||||
bounds accessibleBounds
|
||||
listLength int
|
||||
expectedStart int
|
||||
expectedEnd int
|
||||
}
|
||||
tests := []test{
|
||||
{
|
||||
label: "inaccessible at end",
|
||||
bounds: accessibleBounds{start: 0, end: 3},
|
||||
listLength: 6,
|
||||
expectedStart: 4,
|
||||
expectedEnd: 5,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.label, func(t *testing.T) {
|
||||
start, end := test.bounds.getInaccessibleRange(test.listLength)
|
||||
|
||||
assert.Equal(t, test.expectedStart, start)
|
||||
assert.Equal(t, test.expectedEnd, end)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"fmt"
|
||||
"image"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
@@ -54,12 +53,12 @@ func (s *Server) initPostMetadata() {
|
||||
|
||||
func (a *App) PreparePostListForClient(c request.CTX, originalList *model.PostList) *model.PostList {
|
||||
list := &model.PostList{
|
||||
Posts: make(map[string]*model.Post, len(originalList.Posts)),
|
||||
Order: originalList.Order,
|
||||
NextPostId: originalList.NextPostId,
|
||||
PrevPostId: originalList.PrevPostId,
|
||||
HasNext: originalList.HasNext,
|
||||
HasInaccessiblePosts: originalList.HasInaccessiblePosts,
|
||||
Posts: make(map[string]*model.Post, len(originalList.Posts)),
|
||||
Order: originalList.Order,
|
||||
NextPostId: originalList.NextPostId,
|
||||
PrevPostId: originalList.PrevPostId,
|
||||
HasNext: originalList.HasNext,
|
||||
FirstInaccessiblePostTime: originalList.FirstInaccessiblePostTime,
|
||||
}
|
||||
|
||||
for id, originalPost := range originalList.Posts {
|
||||
@@ -216,7 +215,7 @@ func (a *App) getFileMetadataForPost(post *model.Post, fromMaster bool) ([]*mode
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return a.GetFileInfosForPost(post.Id, fromMaster)
|
||||
return a.GetFileInfosForPost(post.Id, fromMaster, false)
|
||||
}
|
||||
|
||||
func (a *App) getEmojisAndReactionsForPost(post *model.Post) ([]*model.Emoji, []*model.Reaction, *model.AppError) {
|
||||
@@ -606,7 +605,7 @@ func (a *App) getLinkMetadata(c request.CTX, requestURL string, timestamp int64,
|
||||
|
||||
if body != nil {
|
||||
defer func() {
|
||||
io.Copy(ioutil.Discard, body)
|
||||
io.Copy(io.Discard, body)
|
||||
body.Close()
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/dyatlov/go-opengraph/opengraph"
|
||||
ogimage "github.com/dyatlov/go-opengraph/opengraph/types/image"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
@@ -1157,7 +1158,7 @@ func TestGetImagesForPost(t *testing.T) {
|
||||
Type: model.PostEmbedOpengraph,
|
||||
URL: ogURL,
|
||||
Data: &opengraph.OpenGraph{
|
||||
Images: []*opengraph.Image{
|
||||
Images: []*ogimage.Image{
|
||||
{
|
||||
URL: imageURL,
|
||||
},
|
||||
@@ -1211,7 +1212,7 @@ func TestGetImagesForPost(t *testing.T) {
|
||||
Type: model.PostEmbedOpengraph,
|
||||
URL: ogURL,
|
||||
Data: &opengraph.OpenGraph{
|
||||
Images: []*opengraph.Image{
|
||||
Images: []*ogimage.Image{
|
||||
{
|
||||
SecureURL: imageURL,
|
||||
},
|
||||
@@ -1264,7 +1265,7 @@ func TestGetImagesForPost(t *testing.T) {
|
||||
Type: model.PostEmbedOpengraph,
|
||||
URL: ogURL,
|
||||
Data: &opengraph.OpenGraph{
|
||||
Images: []*opengraph.Image{
|
||||
Images: []*ogimage.Image{
|
||||
{
|
||||
URL: server.URL + "/image.png",
|
||||
SecureURL: imageURL,
|
||||
@@ -2711,7 +2712,7 @@ func TestSanitizePostMetadataForUserAndChannel(t *testing.T) {
|
||||
Type: model.PostEmbedOpengraph,
|
||||
URL: "ogURL",
|
||||
Data: &opengraph.OpenGraph{
|
||||
Images: []*opengraph.Image{
|
||||
Images: []*ogimage.Image{
|
||||
{
|
||||
URL: "imageURL",
|
||||
},
|
||||
|
||||
@@ -219,7 +219,7 @@ func TestAttachFilesToPost(t *testing.T) {
|
||||
appErr := th.App.attachFilesToPost(post)
|
||||
assert.Nil(t, appErr)
|
||||
|
||||
infos, appErr := th.App.GetFileInfosForPost(post.Id, false)
|
||||
infos, appErr := th.App.GetFileInfosForPost(post.Id, false, false)
|
||||
assert.Nil(t, appErr)
|
||||
assert.Len(t, infos, 2)
|
||||
})
|
||||
@@ -247,7 +247,7 @@ func TestAttachFilesToPost(t *testing.T) {
|
||||
appErr := th.App.attachFilesToPost(post)
|
||||
assert.Nil(t, appErr)
|
||||
|
||||
infos, appErr := th.App.GetFileInfosForPost(post.Id, false)
|
||||
infos, appErr := th.App.GetFileInfosForPost(post.Id, false, false)
|
||||
assert.Nil(t, appErr)
|
||||
assert.Len(t, infos, 1)
|
||||
assert.Equal(t, info2.Id, infos[0].Id)
|
||||
|
||||
194
app/server.go
194
app/server.go
@@ -9,10 +9,8 @@ import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"hash/maphash"
|
||||
"html/template"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/pprof"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -27,7 +25,6 @@ import (
|
||||
|
||||
"github.com/getsentry/sentry-go"
|
||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/rs/cors"
|
||||
@@ -35,6 +32,7 @@ import (
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/app/email"
|
||||
"github.com/mattermost/mattermost-server/v6/app/featureflag"
|
||||
"github.com/mattermost/mattermost-server/v6/app/platform"
|
||||
"github.com/mattermost/mattermost-server/v6/app/request"
|
||||
"github.com/mattermost/mattermost-server/v6/app/teams"
|
||||
"github.com/mattermost/mattermost-server/v6/app/users"
|
||||
@@ -131,10 +129,6 @@ type Server struct {
|
||||
|
||||
localModeServer *http.Server
|
||||
|
||||
metricsServer *http.Server
|
||||
metricsRouter *mux.Router
|
||||
metricsLock sync.Mutex
|
||||
|
||||
didFinishListen chan struct{}
|
||||
|
||||
goroutineCount int32
|
||||
@@ -177,6 +171,7 @@ type Server struct {
|
||||
configStore *configWrapper
|
||||
filestore filestore.FileBackend
|
||||
|
||||
platformService *platform.PlatformService
|
||||
telemetryService *telemetry.TelemetryService
|
||||
userService *users.UserService
|
||||
teamService *teams.TeamService
|
||||
@@ -256,6 +251,17 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
s.configStore = &configWrapper{srv: s, Store: configStore}
|
||||
}
|
||||
|
||||
ps, sErr := platform.New(platform.ServiceConfig{
|
||||
ConfigStore: s.configStore.Store,
|
||||
StartMetrics: s.startMetrics,
|
||||
Metrics: s.Metrics,
|
||||
Cluster: s.Cluster,
|
||||
})
|
||||
if sErr != nil {
|
||||
return nil, errors.Wrap(sErr, "failed to initialize platform")
|
||||
}
|
||||
s.platformService = ps
|
||||
|
||||
// Step 2: Logging
|
||||
if err := s.initLogging(); err != nil {
|
||||
mlog.Error("Could not initiate logging", mlog.Err(err))
|
||||
@@ -619,10 +625,6 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
s.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableDeveloper = true })
|
||||
}
|
||||
|
||||
if s.startMetrics {
|
||||
s.SetupMetricsServer()
|
||||
}
|
||||
|
||||
s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
|
||||
if (oldLicense == nil && newLicense == nil) || !s.startMetrics {
|
||||
return
|
||||
@@ -632,7 +634,7 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
return
|
||||
}
|
||||
|
||||
s.SetupMetricsServer()
|
||||
s.platformService.RestartMetrics() // TODO: remove when this moved to the platform service
|
||||
})
|
||||
|
||||
s.SearchEngine.UpdateConfig(s.Config())
|
||||
@@ -679,6 +681,7 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
s.runLicenseExpirationCheckJob()
|
||||
s.runInactivityCheckJob()
|
||||
runDNDStatusExpireJob(appInstance)
|
||||
runPostReminderJob(appInstance)
|
||||
})
|
||||
s.runJobs()
|
||||
}
|
||||
@@ -700,24 +703,6 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Server) SetupMetricsServer() {
|
||||
if !*s.Config().MetricsSettings.Enable {
|
||||
return
|
||||
}
|
||||
|
||||
s.StopMetricsServer()
|
||||
|
||||
if err := s.InitMetricsRouter(); err != nil {
|
||||
mlog.Error("Error initiating metrics router.", mlog.Err(err))
|
||||
}
|
||||
|
||||
if s.Metrics != nil {
|
||||
s.Metrics.Register()
|
||||
}
|
||||
|
||||
s.startMetricsServer()
|
||||
}
|
||||
|
||||
func maxInt(a, b int) int {
|
||||
if a > b {
|
||||
return a
|
||||
@@ -1045,7 +1030,7 @@ func (s *Server) Shutdown() {
|
||||
s.Cluster.StopInterNodeCommunication()
|
||||
}
|
||||
|
||||
s.StopMetricsServer()
|
||||
s.platformService.ShutdownMetrics()
|
||||
|
||||
// This must be done after the cluster is stopped.
|
||||
if s.Jobs != nil {
|
||||
@@ -1629,104 +1614,9 @@ func doConfigCleanup(s *Server) {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) StopMetricsServer() {
|
||||
s.metricsLock.Lock()
|
||||
defer s.metricsLock.Unlock()
|
||||
|
||||
if s.metricsServer != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), TimeToWaitForConnectionsToCloseOnServerShutdown)
|
||||
defer cancel()
|
||||
|
||||
s.metricsServer.Shutdown(ctx)
|
||||
s.Log.Info("Metrics and profiling server is stopping")
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: remove this method when we switch to using platform service.
|
||||
func (s *Server) HandleMetrics(route string, h http.Handler) {
|
||||
if s.metricsRouter != nil {
|
||||
s.metricsRouter.Handle(route, h)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) InitMetricsRouter() error {
|
||||
s.metricsRouter = mux.NewRouter()
|
||||
runtime.SetBlockProfileRate(*s.Config().MetricsSettings.BlockProfileRate)
|
||||
|
||||
metricsPage := `
|
||||
<html>
|
||||
<body>{{if .}}
|
||||
<div><a href="/metrics">Metrics</a></div>{{end}}
|
||||
<div><a href="/debug/pprof/">Profiling Root</a></div>
|
||||
<div><a href="/debug/pprof/cmdline">Profiling Command Line</a></div>
|
||||
<div><a href="/debug/pprof/symbol">Profiling Symbols</a></div>
|
||||
<div><a href="/debug/pprof/goroutine">Profiling Goroutines</a></div>
|
||||
<div><a href="/debug/pprof/heap">Profiling Heap</a></div>
|
||||
<div><a href="/debug/pprof/threadcreate">Profiling Threads</a></div>
|
||||
<div><a href="/debug/pprof/block">Profiling Blocking</a></div>
|
||||
<div><a href="/debug/pprof/trace">Profiling Execution Trace</a></div>
|
||||
<div><a href="/debug/pprof/profile">Profiling CPU</a></div>
|
||||
</body>
|
||||
</html>
|
||||
`
|
||||
metricsPageTmpl, err := template.New("page").Parse(metricsPage)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create template")
|
||||
}
|
||||
|
||||
rootHandler := func(w http.ResponseWriter, r *http.Request) {
|
||||
metricsPageTmpl.Execute(w, s.Metrics != nil)
|
||||
}
|
||||
|
||||
s.metricsRouter.HandleFunc("/", rootHandler)
|
||||
s.metricsRouter.StrictSlash(true)
|
||||
|
||||
s.metricsRouter.Handle("/debug", http.RedirectHandler("/", http.StatusMovedPermanently))
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/", pprof.Index)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||
|
||||
// Manually add support for paths linked to by index page at /debug/pprof/
|
||||
s.metricsRouter.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine"))
|
||||
s.metricsRouter.Handle("/debug/pprof/heap", pprof.Handler("heap"))
|
||||
s.metricsRouter.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate"))
|
||||
s.metricsRouter.Handle("/debug/pprof/block", pprof.Handler("block"))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) startMetricsServer() {
|
||||
var notify chan struct{}
|
||||
s.metricsLock.Lock()
|
||||
defer func() {
|
||||
if notify != nil {
|
||||
<-notify
|
||||
}
|
||||
s.metricsLock.Unlock()
|
||||
}()
|
||||
|
||||
l, err := net.Listen("tcp", *s.Config().MetricsSettings.ListenAddress)
|
||||
if err != nil {
|
||||
mlog.Error(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
notify = make(chan struct{})
|
||||
s.metricsServer = &http.Server{
|
||||
Handler: handlers.RecoveryHandler(handlers.PrintRecoveryStack(true))(s.metricsRouter),
|
||||
ReadTimeout: time.Duration(*s.Config().ServiceSettings.ReadTimeout) * time.Second,
|
||||
WriteTimeout: time.Duration(*s.Config().ServiceSettings.WriteTimeout) * time.Second,
|
||||
}
|
||||
|
||||
go func() {
|
||||
close(notify)
|
||||
if err := s.metricsServer.Serve(l); err != nil && err != http.ErrServerClosed {
|
||||
mlog.Critical(err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
s.Log.Info("Metrics and profiling server is started", mlog.String("address", l.Addr().String()))
|
||||
s.platformService.HandleMetrics(route, h)
|
||||
}
|
||||
|
||||
func (s *Server) sendLicenseUpForRenewalEmail(users map[string]*model.User, license *model.License) *model.AppError {
|
||||
@@ -2177,31 +2067,53 @@ func (s *Server) ReadFile(path string) ([]byte, *model.AppError) {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func createDNDStatusExpirationRecurringTask(a *App) {
|
||||
a.ch.dndTaskMut.Lock()
|
||||
a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute)
|
||||
a.ch.dndTaskMut.Unlock()
|
||||
func withMut(mut *sync.Mutex, f func()) {
|
||||
mut.Lock()
|
||||
defer mut.Unlock()
|
||||
f()
|
||||
}
|
||||
|
||||
func cancelDNDStatusExpirationRecurringTask(a *App) {
|
||||
a.ch.dndTaskMut.Lock()
|
||||
if a.ch.dndTask != nil {
|
||||
a.ch.dndTask.Cancel()
|
||||
a.ch.dndTask = nil
|
||||
func cancelTask(mut *sync.Mutex, task *model.ScheduledTask) {
|
||||
mut.Lock()
|
||||
defer mut.Unlock()
|
||||
if task != nil {
|
||||
task.Cancel()
|
||||
task = nil
|
||||
}
|
||||
a.ch.dndTaskMut.Unlock()
|
||||
}
|
||||
|
||||
func runDNDStatusExpireJob(a *App) {
|
||||
if a.IsLeader() {
|
||||
createDNDStatusExpirationRecurringTask(a)
|
||||
withMut(&a.ch.dndTaskMut, func() {
|
||||
a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute)
|
||||
})
|
||||
}
|
||||
a.ch.srv.AddClusterLeaderChangedListener(func() {
|
||||
mlog.Info("Cluster leader changed. Determining if unset DNS status task should be running", mlog.Bool("isLeader", a.IsLeader()))
|
||||
if a.IsLeader() {
|
||||
createDNDStatusExpirationRecurringTask(a)
|
||||
withMut(&a.ch.dndTaskMut, func() {
|
||||
a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute)
|
||||
})
|
||||
} else {
|
||||
cancelDNDStatusExpirationRecurringTask(a)
|
||||
cancelTask(&a.ch.dndTaskMut, a.ch.dndTask)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func runPostReminderJob(a *App) {
|
||||
if a.IsLeader() {
|
||||
withMut(&a.ch.postReminderMut, func() {
|
||||
a.ch.postReminderTask = model.CreateRecurringTaskFromNextIntervalTime("Check Post reminders", a.CheckPostReminders, 5*time.Minute)
|
||||
})
|
||||
}
|
||||
a.ch.srv.AddClusterLeaderChangedListener(func() {
|
||||
mlog.Info("Cluster leader changed. Determining if post reminder task should be running", mlog.Bool("isLeader", a.IsLeader()))
|
||||
if a.IsLeader() {
|
||||
withMut(&a.ch.postReminderMut, func() {
|
||||
a.ch.postReminderTask = model.CreateRecurringTaskFromNextIntervalTime("Check Post reminders", a.CheckPostReminders, 5*time.Minute)
|
||||
})
|
||||
} else {
|
||||
cancelTask(&a.ch.postReminderMut, a.ch.postReminderTask)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -6,7 +6,6 @@ package slashcommands
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"path"
|
||||
"regexp"
|
||||
@@ -507,7 +506,7 @@ func (*LoadTestProvider) URLCommand(a *app.App, c *request.Context, args *model.
|
||||
return &model.CommandResponse{Text: "Unable to get file", ResponseType: model.CommandResponseTypeEphemeral}, err
|
||||
}
|
||||
defer func() {
|
||||
io.Copy(ioutil.Discard, r.Body)
|
||||
io.Copy(io.Discard, r.Body)
|
||||
r.Body.Close()
|
||||
}()
|
||||
|
||||
@@ -565,7 +564,7 @@ func (*LoadTestProvider) JsonCommand(a *app.App, c *request.Context, args *model
|
||||
return &model.CommandResponse{Text: "Unable to get file", ResponseType: model.CommandResponseTypeEphemeral}, errors.Errorf("unexpected status code %d", r.StatusCode)
|
||||
}
|
||||
defer func() {
|
||||
io.Copy(ioutil.Discard, r.Body)
|
||||
io.Copy(io.Discard, r.Body)
|
||||
r.Body.Close()
|
||||
}()
|
||||
|
||||
|
||||
@@ -24,10 +24,19 @@ const minFirstPartSize = 5 * 1024 * 1024 // 5MB
|
||||
|
||||
func (a *App) genFileInfoFromReader(name string, file io.ReadSeeker, size int64) (*model.FileInfo, error) {
|
||||
ext := strings.ToLower(filepath.Ext(name))
|
||||
|
||||
info := &model.FileInfo{
|
||||
Name: name,
|
||||
MimeType: mime.TypeByExtension(ext),
|
||||
Name: name,
|
||||
MimeType: mime.TypeByExtension(ext),
|
||||
Size: size,
|
||||
Extension: ext,
|
||||
}
|
||||
|
||||
if ext != "" {
|
||||
// The client expects a file extension without the leading period
|
||||
info.Extension = ext[1:]
|
||||
}
|
||||
|
||||
if info.IsImage() {
|
||||
config, _, err := a.ch.imgDecoder.DecodeConfig(file)
|
||||
if err != nil {
|
||||
|
||||
Ссылка в новой задаче
Block a user