autobindings/experiments/server_hosting/servers_functionality.go

355 lines
10 KiB
Go

package server_hosting
import (
"cwtch.im/cwtch/app"
"cwtch.im/cwtch/event"
"git.openprivacy.ca/cwtch.im/cwtch-autobindings/constants"
"git.openprivacy.ca/cwtch.im/server"
"git.openprivacy.ca/openprivacy/connectivity"
"git.openprivacy.ca/openprivacy/log"
"os"
"path/filepath"
"strconv"
"sync"
"time"
)
const serversExperiment = "servers-experiment"
const (
ZeroServersLoaded = event.Type("ZeroServersLoaded")
NewServer = event.Type("NewServer")
ServerIntentUpdate = event.Type("ServerIntentUpdate")
ServerDeleted = event.Type("ServerDeleted")
ServerStatsUpdate = event.Type("ServerStatsUpdate")
)
const (
Intent = event.Field("Intent")
TotalMessages = event.Field("TotalMessages")
Connections = event.Field("Connections")
)
const (
IntentRunning = "running"
IntentStopped = "stopped"
)
type ServerInfo struct {
Onion string
ServerBundle string
Autostart bool
Running bool
Description string
StorageType string
}
var lock sync.Mutex
var appServers server.Servers
var killStatsUpdate chan bool = make(chan bool, 1)
var enabled bool = false
func Init(acn connectivity.ACN, appdir string) *ServersFunctionality {
lock.Lock()
defer lock.Unlock()
if appServers == nil {
serversDir := filepath.Join(appdir, "servers")
err := os.MkdirAll(serversDir, 0700)
if err != nil {
log.Errorf("Could not init servers directory: %s", err)
}
appServers = server.NewServers(acn, serversDir)
}
serversFunctionality := new(ServersFunctionality)
return serversFunctionality
}
// track acnStatus across events
var acnStatus = -1
func (sf *ServersFunctionality) OnACNStatusEvent(appl app.Application, e *event.Event) {
newAcnStatus, err := strconv.Atoi(e.Data[event.Progress])
if !appl.IsFeatureEnabled(serversExperiment) || err != nil {
return
}
if newAcnStatus == 100 {
if acnStatus != 100 {
// just came online
for _, onion := range sf.ListServers() {
autostart := false
if s := sf.GetServer(onion); s != nil {
autostart = s.GetAttribute(server.AttrAutostart) == "true"
}
if autostart {
sf.LaunchServer(appl, onion)
}
}
}
} else {
if acnStatus == 100 {
for _, onion := range sf.ListServers() {
sf.StopServer(appl, onion)
}
}
}
acnStatus = newAcnStatus
}
func (sh *ServersFunctionality) Disable() {
lock.Lock()
defer lock.Unlock()
if appServers != nil {
appServers.Stop()
}
if enabled {
enabled = false
killStatsUpdate <- true
}
}
func (sh *ServersFunctionality) Enabled() bool {
lock.Lock()
defer lock.Unlock()
return enabled
}
// ServersFunctionality provides experiment gated server functionality
type ServersFunctionality struct {
}
func (sh *ServersFunctionality) UpdateSettings(appl app.Application, acn connectivity.ACN) {
if appl.IsFeatureEnabled(serversExperiment) {
sh.LoadServers(appl, acn, app.DefactoPasswordForUnencryptedProfiles)
if !sh.Enabled() {
sh.Enable(appl, acn)
// republish servers
serversList := sh.ListServers()
for _, server := range serversList {
serverInfo := sh.GetServerInfo(server)
ev := event.NewEvent(NewServer, make(map[event.Field]string))
serverInfo.EnrichEvent(&ev)
appl.GetPrimaryBus().Publish(ev)
}
sh.LaunchServers(appl, acn)
}
} else {
sh.Disable()
}
}
func (sh *ServersFunctionality) SetServerAttribute(appl app.Application, handle string, key string, val string) {
if appl.IsFeatureEnabled(serversExperiment) {
server := sh.GetServer(handle)
if server != nil {
server.SetAttribute(key, val)
}
}
}
func (sh *ServersFunctionality) StopServers(appl app.Application) {
// we are always permitted to stop servers
for _, onion := range sh.ListServers() {
sh.StopServer(appl, onion)
}
}
func (sh *ServersFunctionality) LaunchServers(appl app.Application, acn connectivity.ACN) {
if appl.IsFeatureEnabled(serversExperiment) {
acnStatus, _ := acn.GetBootstrapStatus()
if acnStatus == 100 {
for _, onion := range sh.ListServers() {
autostart := false
if s := sh.GetServer(onion); s != nil {
autostart = s.GetAttribute(server.AttrAutostart) == "true"
}
if autostart {
sh.LaunchServer(appl, onion)
}
}
}
}
}
func (sf *ServersFunctionality) Enable(application app.Application, acn connectivity.ACN) {
lock.Lock()
defer lock.Unlock()
if appServers != nil && !enabled {
enabled = true
// load unencrypted profiles
sf.LoadServers(application, acn, app.DefactoPasswordForUnencryptedProfiles)
go cacheForwardServerMetricUpdates(application)
}
}
func (sf *ServersFunctionality) LoadServers(appl app.Application, acn connectivity.ACN, password string) {
serversList, err := appServers.LoadServers(password)
if err != nil {
log.Errorf("Error attempting to load servers :%s\n", err)
appl.GetPrimaryBus().Publish(event.NewEventList(ZeroServersLoaded))
} else if len(serversList) == 0 {
log.Debugln("Loaded 0 servers")
appl.GetPrimaryBus().Publish(event.NewEventList(ZeroServersLoaded))
} else {
acnStatus, _ := acn.GetBootstrapStatus()
for _, serverOnion := range serversList {
serverInfo := sf.GetServerInfo(serverOnion)
log.Debugf("Load Server NewServer event: %s", serverInfo)
ev := event.NewEvent(NewServer, make(map[event.Field]string))
serverInfo.EnrichEvent(&ev)
appl.GetPrimaryBus().Publish(ev)
if serverInfo.Autostart && acnStatus == 100 {
sf.LaunchServer(appl, serverOnion)
}
}
}
// server:1.3/libcwtch-go:1.4 accidentally enabled monitor logging by default. make sure it's turned off
for _, onion := range serversList {
server := appServers.GetServer(onion)
server.SetMonitorLogging(false)
}
}
func (sf *ServersFunctionality) CreateServer(appl app.Application, password string, description string, autostart bool) {
s, err := appServers.CreateServer(password)
if err != nil {
log.Errorf("Could not create new server: %s\n", err)
} else {
s.SetAttribute(server.AttrDescription, description)
if autostart {
s.SetAttribute(server.AttrAutostart, "true")
} else {
s.SetAttribute(server.AttrAutostart, "false")
}
if password == app.DefactoPasswordForUnencryptedProfiles {
s.SetAttribute(server.AttrStorageType, server.StorageTypeDefaultPassword)
} else {
s.SetAttribute(server.AttrStorageType, server.StorageTypePassword)
}
serverInfo := sf.GetServerInfo(s.Onion())
log.Debugf("Creating Server NewServer event: %s", serverInfo)
ev := event.NewEvent(NewServer, make(map[event.Field]string))
serverInfo.EnrichEvent(&ev)
appl.GetPrimaryBus().Publish(ev)
if autostart {
sf.LaunchServer(appl, s.Onion())
}
}
}
func (sf *ServersFunctionality) GetServer(onion string) server.Server {
return appServers.GetServer(onion)
}
func (sf *ServersFunctionality) GetServerStatistics(onion string) server.Statistics {
s := appServers.GetServer(onion)
if s != nil {
return s.GetStatistics()
}
return server.Statistics{}
}
func (sf *ServersFunctionality) ListServers() []string {
return appServers.ListServers()
}
func (sf *ServersFunctionality) DeleteServer(appl app.Application, onion string, currentPassword string) error {
sf.StopServer(appl, onion)
appl.GetPrimaryBus().Publish(event.NewEventList(ServerIntentUpdate, event.Identity, onion, Intent, IntentStopped))
err := appServers.DeleteServer(onion, currentPassword)
if err == nil {
appl.GetPrimaryBus().Publish(event.NewEventList(ServerDeleted, event.Status, constants.StatusSuccess, event.Identity, onion))
} else {
appl.GetPrimaryBus().Publish(event.NewEventList(ServerDeleted, event.Status, constants.StatusError, event.Error, err.Error(), event.Identity, onion))
}
return err
}
func (sf *ServersFunctionality) LaunchServer(appl app.Application, onion string) {
if appl.IsFeatureEnabled(serversExperiment) {
appServers.LaunchServer(onion)
server := appServers.GetServer(onion)
if server != nil {
newStats := server.GetStatistics()
appl.GetPrimaryBus().Publish(event.NewEventList(ServerStatsUpdate, event.Identity, onion, TotalMessages, strconv.Itoa(newStats.TotalMessages), Connections, strconv.Itoa(newStats.TotalConnections)))
appl.GetPrimaryBus().Publish(event.NewEventList(ServerIntentUpdate, event.Identity, onion, Intent, IntentRunning))
}
}
}
func (sf *ServersFunctionality) StopServer(appl app.Application, onion string) {
// we are always permitted to stop servers
appServers.StopServer(onion)
appl.GetPrimaryBus().Publish(event.NewEventList(ServerIntentUpdate, event.Identity, onion, Intent, IntentStopped))
}
func (sf *ServersFunctionality) DestroyServers() {
// we are always permitted to destroy servers
appServers.Destroy()
}
func (sf *ServersFunctionality) GetServerInfo(onion string) *ServerInfo {
s := sf.GetServer(onion)
var serverInfo ServerInfo
serverInfo.Onion = s.Onion()
serverInfo.ServerBundle = s.ServerBundle()
serverInfo.Autostart = s.GetAttribute(server.AttrAutostart) == "true"
running, _ := s.CheckStatus()
serverInfo.Running = running
serverInfo.Description = s.GetAttribute(server.AttrDescription)
serverInfo.StorageType = s.GetAttribute(server.AttrStorageType)
return &serverInfo
}
func (si *ServerInfo) EnrichEvent(e *event.Event) {
e.Data["Onion"] = si.Onion
e.Data["ServerBundle"] = si.ServerBundle
e.Data["Description"] = si.Description
e.Data["StorageType"] = si.StorageType
if si.Autostart {
e.Data["Autostart"] = "true"
} else {
e.Data["Autostart"] = "false"
}
if si.Running {
e.Data["Running"] = "true"
} else {
e.Data["Running"] = "false"
}
}
// cacheForwardServerMetricUpdates every minute gets metrics for all servers, and if they have changed, sends events to the UI
func cacheForwardServerMetricUpdates(appl app.Application) {
var cache map[string]server.Statistics = make(map[string]server.Statistics)
duration := time.Second // allow first load
for {
select {
case <-time.After(duration):
duration = time.Minute
serverList := appServers.ListServers()
for _, serverOnion := range serverList {
server := appServers.GetServer(serverOnion)
if running, err := server.CheckStatus(); running && err == nil {
newStats := server.GetStatistics()
if stats, ok := cache[serverOnion]; !ok || stats.TotalConnections != newStats.TotalConnections || stats.TotalMessages != newStats.TotalMessages {
cache[serverOnion] = newStats
appl.GetPrimaryBus().Publish(event.NewEventList(ServerStatsUpdate, event.Identity, serverOnion, TotalMessages, strconv.Itoa(newStats.TotalMessages), Connections, strconv.Itoa(newStats.TotalConnections)))
}
}
}
case <-killStatsUpdate:
return
}
}
}
func (sh *ServersFunctionality) Shutdown() {
sh.Disable()
appServers.Destroy()
}