PLT-2899 adding clustering of app servers (#3682)

* PLT-2899 adding clustering of app servers

* PLT-2899 base framework

* PLT-2899 HA backend

* PLT-2899 Fixing config file

* PLT-2899 adding config syncing

* PLT-2899 set System console to readonly when clustering enabled.

* PLT-2899 Fixing publish API

* PLT-2899 fixing strings

api/admin.go

@@ -46,6 +46,7 @@ func InitAdmin() {
 	BaseRoutes.Admin.Handle("/add_certificate", ApiAdminSystemRequired(addCertificate)).Methods("POST")
 	BaseRoutes.Admin.Handle("/remove_certificate", ApiAdminSystemRequired(removeCertificate)).Methods("POST")
 	BaseRoutes.Admin.Handle("/saml_cert_status", ApiAdminSystemRequired(samlCertificateStatus)).Methods("GET")
+	BaseRoutes.Admin.Handle("/cluster_status", ApiAdminSystemRequired(getClusterStatus)).Methods("GET")
 }
 
 func getLogs(c *Context, w http.ResponseWriter, r *http.Request) {
@@ -54,13 +55,32 @@ func getLogs(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	lines, err := GetLogs()
+	if err != nil {
+		c.Err = err
+		return
+	}
+
+	if einterfaces.GetClusterInterface() != nil {
+		clines, err := einterfaces.GetClusterInterface().GetLogs()
+		if err != nil {
+			c.Err = err
+			return
+		}
+
+		lines = append(lines, clines...)
+	}
+
+	w.Write([]byte(model.ArrayToJson(lines)))
+}
+
+func GetLogs() ([]string, *model.AppError) {
 	var lines []string
 
 	if utils.Cfg.LogSettings.EnableFile {
-
 		file, err := os.Open(utils.GetLogFileLocation(utils.Cfg.LogSettings.FileLocation))
 		if err != nil {
-			c.Err = model.NewLocAppError("getLogs", "api.admin.file_read_error", nil, err.Error())
+			return nil, model.NewLocAppError("getLogs", "api.admin.file_read_error", nil, err.Error())
 		}
 
 		defer file.Close()
@@ -73,7 +93,21 @@ func getLogs(c *Context, w http.ResponseWriter, r *http.Request) {
 		lines = append(lines, "")
 	}
 
-	w.Write([]byte(model.ArrayToJson(lines)))
+	return lines, nil
+}
+
+func getClusterStatus(c *Context, w http.ResponseWriter, r *http.Request) {
+
+	if !c.HasSystemAdminPermissions("getClusterStatus") {
+		return
+	}
+
+	infos := make([]*model.ClusterInfo, 0)
+	if einterfaces.GetClusterInterface() != nil {
+		infos = einterfaces.GetClusterInterface().GetClusterInfos()
+	}
+
+	w.Write([]byte(model.ClusterInfosToJson(infos)))
 }
 
 func getAllAudits(c *Context, w http.ResponseWriter, r *http.Request) {
@@ -150,11 +184,26 @@ func saveConfig(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if *utils.Cfg.ClusterSettings.Enable {
+		c.Err = model.NewLocAppError("saveConfig", "ent.cluster.save_config.error", nil, "")
+		return
+	}
+
 	c.LogAudit("")
 
+	//oldCfg := utils.Cfg
 	utils.SaveConfig(utils.CfgFileName, cfg)
 	utils.LoadConfig(utils.CfgFileName)
 
+	// Future feature is to sync the configuration files
+	// if einterfaces.GetClusterInterface() != nil {
+	// 	err := einterfaces.GetClusterInterface().ConfigChanged(cfg, oldCfg, true)
+	// 	if err != nil {
+	// 		c.Err = err
+	// 		return
+	// 	}
+	// }
+
 	rdata := map[string]string{}
 	rdata["status"] = "OK"
 	w.Write([]byte(model.MapToJson(rdata)))

api/admin_test.go

@@ -25,6 +25,18 @@ func TestGetLogs(t *testing.T) {
 	}
 }
 
+func TestGetClusterInfos(t *testing.T) {
+	th := Setup().InitSystemAdmin().InitBasic()
+
+	if _, err := th.BasicClient.GetClusterStatus(); err == nil {
+		t.Fatal("Shouldn't have permissions")
+	}
+
+	if _, err := th.SystemAdminClient.GetClusterStatus(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestGetAllAudits(t *testing.T) {
 	th := Setup().InitBasic().InitSystemAdmin()
 

api/context.go

@@ -14,13 +14,13 @@ import (
 	"github.com/gorilla/mux"
 	goi18n "github.com/nicksnyder/go-i18n/i18n"
 
+	"github.com/mattermost/platform/einterfaces"
 	"github.com/mattermost/platform/model"
 	"github.com/mattermost/platform/store"
 	"github.com/mattermost/platform/utils"
 )
 
 var sessionCache *utils.Cache = utils.NewLru(model.SESSION_CACHE_SIZE)
-var statusCache *utils.Cache = utils.NewLru(model.STATUS_CACHE_SIZE)
 
 var allowedMethods []string = []string{
 	"POST",
@@ -148,7 +148,10 @@ func (h handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	w.Header().Set(model.HEADER_REQUEST_ID, c.RequestId)
-	w.Header().Set(model.HEADER_VERSION_ID, fmt.Sprintf("%v.%v", model.CurrentVersion, utils.CfgLastModified))
+	w.Header().Set(model.HEADER_VERSION_ID, fmt.Sprintf("%v.%v", model.CurrentVersion, utils.CfgHash))
+	if einterfaces.GetClusterInterface() != nil {
+		w.Header().Set(model.HEADER_CLUSTER_ID, einterfaces.GetClusterInterface().GetClusterId())
+	}
 
 	// Instruct the browser not to display us in an iframe unless is the same origin for anti-clickjacking
 	if !h.isApi {
@@ -554,6 +557,10 @@ func RemoveAllSessionsForUserId(userId string) {
 			}
 		}
 	}
+
+	if einterfaces.GetClusterInterface() != nil {
+		einterfaces.GetClusterInterface().RemoveAllSessionsForUserId(userId)
+	}
 }
 
 func AddSessionToCache(session *model.Session) {

api/general.go

@@ -69,7 +69,6 @@ func ping(c *Context, w http.ResponseWriter, r *http.Request) {
 	m := make(map[string]string)
 	m["version"] = model.CurrentVersion
 	m["server_time"] = fmt.Sprintf("%v", model.GetMillis())
-	m["node_id"] = ""
 	w.Write([]byte(model.MapToJson(m)))
 }
 

api/status.go

@@ -7,11 +7,23 @@ import (
 	"net/http"
 
 	l4g "github.com/alecthomas/log4go"
+
+	"github.com/mattermost/platform/einterfaces"
 	"github.com/mattermost/platform/model"
 	"github.com/mattermost/platform/store"
 	"github.com/mattermost/platform/utils"
 )
 
+var statusCache *utils.Cache = utils.NewLru(model.STATUS_CACHE_SIZE)
+
+func AddStatusCache(status *model.Status) {
+	statusCache.Add(status.UserId, status)
+
+	if einterfaces.GetClusterInterface() != nil {
+		einterfaces.GetClusterInterface().UpdateStatus(status)
+	}
+}
+
 func InitStatus() {
 	l4g.Debug(utils.T("api.status.init.debug"))
 
@@ -69,7 +81,7 @@ func SetStatusOnline(userId string, sessionId string) {
 		status.LastActivityAt = model.GetMillis()
 	}
 
-	statusCache.Add(status.UserId, status)
+	AddStatusCache(status)
 
 	achan := Srv.Store.Session().UpdateLastActivityAt(sessionId, model.GetMillis())
 
@@ -98,7 +110,7 @@ func SetStatusOnline(userId string, sessionId string) {
 func SetStatusOffline(userId string) {
 	status := &model.Status{userId, model.STATUS_OFFLINE, model.GetMillis()}
 
-	statusCache.Add(status.UserId, status)
+	AddStatusCache(status)
 
 	if result := <-Srv.Store.Status().SaveOrUpdate(status); result.Err != nil {
 		l4g.Error(utils.T("api.status.save_status.error"), userId, result.Err)
@@ -125,7 +137,7 @@ func SetStatusAwayIfNeeded(userId string) {
 
 	status.Status = model.STATUS_AWAY
 
-	statusCache.Add(status.UserId, status)
+	AddStatusCache(status)
 
 	if result := <-Srv.Store.Status().SaveOrUpdate(status); result.Err != nil {
 		l4g.Error(utils.T("api.status.save_status.error"), userId, result.Err)

api/web_hub.go

@@ -5,6 +5,8 @@ package api
 
 import (
 	l4g "github.com/alecthomas/log4go"
+
+	"github.com/mattermost/platform/einterfaces"
 	"github.com/mattermost/platform/model"
 	"github.com/mattermost/platform/utils"
 )
@@ -31,14 +33,30 @@ var hub = &Hub{
 
 func Publish(message *model.WebSocketEvent) {
 	hub.Broadcast(message)
+
+	if einterfaces.GetClusterInterface() != nil {
+		einterfaces.GetClusterInterface().Publish(message)
+	}
+}
+
+func PublishSkipClusterSend(message *model.WebSocketEvent) {
+	hub.Broadcast(message)
 }
 
 func InvalidateCacheForUser(userId string) {
 	hub.invalidateUser <- userId
+
+	if einterfaces.GetClusterInterface() != nil {
+		einterfaces.GetClusterInterface().InvalidateCacheForUser(userId)
+	}
 }
 
 func InvalidateCacheForChannel(channelId string) {
 	hub.invalidateChannel <- channelId
+
+	if einterfaces.GetClusterInterface() != nil {
+		einterfaces.GetClusterInterface().InvalidateCacheForChannel(channelId)
+	}
 }
 
 func (h *Hub) Register(webConn *WebConn) {