spdf/elektron
Archived
1
0
Fork 0
Code Issues 4 Pull requests 1 Projects Releases Packages Wiki Activity

Changes made everywhere inorder to use elektronLogging library for logging

Browse source
This commit is contained in:
balandi1 2019-11-13 14:08:52 -05:00
parent b501054412
commit c27aba895b
23 changed files with 382 additions and 308 deletions
Download patch file Download diff file Expand all files Collapse all files

6
elektronLogging/ElektronFormatter.go
View file

@ -37,8 +37,10 @@ func (f ElektronFormatter) Format(entry *log.Entry) ([]byte, error) {
levelColor := f.getColor(entry) levelColor := f.getColor(entry)
level := levelColor.Sprintf("[%s]:",strings.ToUpper(entry.Level.String())) level := levelColor.Sprintf("[%s]:",strings.ToUpper(entry.Level.String()))
message := fmt.Sprintf("%s %s %s ",level,entry.Time.Format(f.TimestampFormat), entry.Message) message := fmt.Sprintf("%s %s ",level,entry.Time.Format(f.TimestampFormat))
if entry.Message != "" {
message = fmt.Sprintf("%s %s %s ",level,entry.Time.Format(f.TimestampFormat), entry.Message)
}
var formattedFields []string var formattedFields []string
for key, value := range entry.Data { for key, value := range entry.Data {
formattedFields = append(formattedFields, formattedFields = append(formattedFields,

20
elektronLogging/clsfnTaskDistOverheadLogger.go
View file

@ -1,10 +1,8 @@
package elektronLogging package elektronLogging
import ( import (
//"fmt"
"os" "os"
logrus "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
data "gitlab.com/spdf/elektron/elektronLogging/data"
) )
type ClsfnTaskDistOverheadLogger struct { type ClsfnTaskDistOverheadLogger struct {
@ -18,20 +16,18 @@ func NewClsfnTaskDistOverheadLogger(logType int, prefix string) *ClsfnTaskDistOv
return cLog return cLog
} }
func (cLog *ClsfnTaskDistOverheadLogger) Log(logType int, level logrus.Level, logData data.LogData,message string) { func (cLog *ClsfnTaskDistOverheadLogger) Log(logType int, level log.Level, logData log.Fields,message string) {
if cLog.Type == logType { if cLog.Type == logType {
logFields := cloneFields(logData) logger.SetLevel(level)
log.SetLevel(level)
if cLog.AllowOnConsole { if cLog.AllowOnConsole {
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logFields).Println(message) logger.WithFields(logData).Println(message)
} }
log.SetOutput(cLog.LogFileName) logger.SetOutput(cLog.LogFileName)
log.WithFields(logFields).Println(message) logger.WithFields(logData).Println(message)
} }
if cLog.next != nil { if cLog.next != nil {
cLog.next.Log(logType, level, logData, message) cLog.next.Log(logType, level, logData, message)
@ -45,7 +41,7 @@ func (cLog *ClsfnTaskDistOverheadLogger) SetLogFile(prefix string) {
tskDistLogPrefix = logDir + "/" + tskDistLogPrefix tskDistLogPrefix = logDir + "/" + tskDistLogPrefix
} }
if logFile, err := os.Create(tskDistLogPrefix); err != nil { if logFile, err := os.Create(tskDistLogPrefix); err != nil {
logrus.Fatal("Unable to create logFile: ", err) log.Fatal("Unable to create logFile: ", err)
} else { } else {
cLog.LogFileName = logFile cLog.LogFileName = logFile
cLog.AllowOnConsole = config.TaskDistConfig.AllowOnConsole cLog.AllowOnConsole = config.TaskDistConfig.AllowOnConsole

11
elektronLogging/consoleLogger.go
View file

@ -3,7 +3,6 @@ package elektronLogging
import ( import (
"os" "os"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
//data "github.com/spdfg/elektron/elektronLogging/data"
) )
type ConsoleLogger struct { type ConsoleLogger struct {
@ -20,13 +19,13 @@ func (cLog *ConsoleLogger) Log(logType int, level log.Level, logData log.Fields,
if logType <= cLog.Type { if logType <= cLog.Type {
//logFields := cloneFields(logData) //logFields := cloneFields(logData)
log.SetLevel(level) logger.SetLevel(level)
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
log.SetOutput(cLog.LogFileName) logger.SetOutput(cLog.LogFileName)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
if cLog.next != nil { if cLog.next != nil {
cLog.next.Log(logType, level, logData, message) cLog.next.Log(logType, level, logData, message)

11
elektronLogging/logger.go
View file

@ -11,21 +11,25 @@ import (
var config LoggerConfig var config LoggerConfig
var logger *log.Logger var logger *log.Logger
var formatter ElektronFormatter var formatter ElektronFormatter
//var logDir string
func BuildLogger() *LoggerImpl { func BuildLogger() *LoggerImpl {
// read configuration from yaml
config.GetConfig() config.GetConfig()
// create the log directory
startTime := time.Now() startTime := time.Now()
formatter.TimestampFormat = "2006-01-02 15:04:05" formatter.TimestampFormat = "2006-01-02 15:04:05"
GetLogDir(startTime, "_") GetLogDir(startTime, "_")
prefix := fmt.Sprintf("_%s%s%s%s%s",startTime.Month().String(),strconv.Itoa(startTime.Day()), prefix := fmt.Sprintf("_%d%d%s%s%s%s",startTime.Year(), startTime.Month(),strconv.Itoa(startTime.Day()),
strconv.Itoa(startTime.Hour()),strconv.Itoa(startTime.Minute()),strconv.Itoa(startTime.Second())) strconv.Itoa(startTime.Hour()),strconv.Itoa(startTime.Minute()),strconv.Itoa(startTime.Second()))
//create a single logrus instance and set its formatter to ElektronFormatter
logger = log.New() logger = log.New()
logger.SetFormatter(&formatter) logger.SetFormatter(&formatter)
// create a chain of loggers
head := new(LoggerImpl) head := new(LoggerImpl)
cLog := NewConsoleLogger(CONSOLE,prefix) cLog := NewConsoleLogger(CONSOLE,prefix)
pLog := NewPcpLogger(PCP,prefix) pLog := NewPcpLogger(PCP,prefix)
@ -38,7 +42,6 @@ func BuildLogger() *LoggerImpl {
cLog.SetNext(pLog) cLog.SetNext(pLog)
pLog.SetNext(schedTraceLog) pLog.SetNext(schedTraceLog)
schedTraceLog.SetNext(spsLog) schedTraceLog.SetNext(spsLog)
spsLog.SetNext(schedWindowLog) spsLog.SetNext(schedWindowLog)
schedWindowLog.SetNext(tskDistLog) schedWindowLog.SetNext(tskDistLog)

2
elektronLogging/loggerConfig.go
View file

@ -3,7 +3,7 @@ import (
"gopkg.in/yaml.v2" "gopkg.in/yaml.v2"
"io/ioutil" "io/ioutil"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
elekEnv "gitlab.com/spdf/elektron/environment" elekEnv "github.com/spdfg/elektron/environment"
) )
type LoggerConfig struct { type LoggerConfig struct {

13
elektronLogging/pcpLogger.go
View file

@ -3,7 +3,6 @@ package elektronLogging
import ( import (
"os" "os"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
//data "github.com/spdfg/elektron/elektronLogging/data"
) )
type PcpLogger struct { type PcpLogger struct {
@ -20,17 +19,15 @@ func NewPcpLogger(logType int, prefix string) *PcpLogger {
func (pLog *PcpLogger) Log(logType int, level log.Level, logData log.Fields, message string) { func (pLog *PcpLogger) Log(logType int, level log.Level, logData log.Fields, message string) {
if pLog.Type == logType { if pLog.Type == logType {
//logFields := cloneFields(logData) logger.SetLevel(level)
log.SetLevel(level)
if pLog.AllowOnConsole { if pLog.AllowOnConsole {
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
log.SetOutput(pLog.LogFileName) logger.SetOutput(pLog.LogFileName)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
if pLog.next != nil { if pLog.next != nil {
pLog.next.Log(logType, level, logData, message) pLog.next.Log(logType, level, logData, message)

14
elektronLogging/schedPolicySwitchLogger.go
View file

@ -1,10 +1,8 @@
package elektronLogging package elektronLogging
import ( import (
//"fmt"
"os" "os"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
//data "github.com/spdfg/elektron/elektronLogging/data"
) )
type SchedPolicySwitchLogger struct { type SchedPolicySwitchLogger struct {
@ -21,17 +19,15 @@ func NewSchedPolicySwitchLogger(logType int, prefix string) *SchedPolicySwitchLo
func (sLog *SchedPolicySwitchLogger) Log(logType int, level log.Level, logData log.Fields, message string) { func (sLog *SchedPolicySwitchLogger) Log(logType int, level log.Level, logData log.Fields, message string) {
if sLog.Type == logType { if sLog.Type == logType {
//logFields := cloneFields(logData) logger.SetLevel(level)
log.SetLevel(level)
if sLog.AllowOnConsole { if sLog.AllowOnConsole {
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
log.SetOutput(sLog.LogFileName) logger.SetOutput(sLog.LogFileName)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
if sLog.next != nil { if sLog.next != nil {
sLog.next.Log(logType, level, logData, message) sLog.next.Log(logType, level, logData, message)

14
elektronLogging/schedTraceLogger.go
View file

@ -1,10 +1,8 @@
package elektronLogging package elektronLogging
import ( import (
//"fmt"
"os" "os"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
//data "github.com/spdfg/elektron/elektronLogging/data"
) )
type SchedTraceLogger struct { type SchedTraceLogger struct {
@ -21,17 +19,15 @@ func NewSchedTraceLogger(logType int, prefix string) *SchedTraceLogger {
func (sLog *SchedTraceLogger) Log(logType int, level log.Level, logData log.Fields, message string) { func (sLog *SchedTraceLogger) Log(logType int, level log.Level, logData log.Fields, message string) {
if sLog.Type == logType { if sLog.Type == logType {
//logFields := cloneFields(logData) logger.SetLevel(level)
log.SetLevel(level)
if sLog.AllowOnConsole { if sLog.AllowOnConsole {
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
log.SetOutput(sLog.LogFileName) logger.SetOutput(sLog.LogFileName)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
if sLog.next != nil { if sLog.next != nil {
sLog.next.Log(logType, level, logData, message) sLog.next.Log(logType, level, logData, message)

13
elektronLogging/schedWindowLogger.go
View file

@ -3,7 +3,6 @@ package elektronLogging
import ( import (
"os" "os"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
//data "github.com/spdfg/elektron/elektronLogging/data"
) )
type SchedWindowLogger struct { type SchedWindowLogger struct {
@ -20,16 +19,14 @@ func NewSchedWindowLogger(logType int, prefix string) *SchedWindowLogger {
func (sLog *SchedWindowLogger) Log(logType int, level log.Level, logData log.Fields, message string) { func (sLog *SchedWindowLogger) Log(logType int, level log.Level, logData log.Fields, message string) {
if sLog.Type == logType { if sLog.Type == logType {
//logFields := cloneFields(logData) logger.SetLevel(level)
log.SetLevel(level)
if sLog.AllowOnConsole { if sLog.AllowOnConsole {
log.SetOutput(os.Stdout) logger.SetOutput(os.Stdout)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
log.SetOutput(sLog.LogFileName) logger.SetOutput(sLog.LogFileName)
log.WithFields(logData).Println(message) logger.WithFields(logData).Println(message)
} }
if sLog.next != nil { if sLog.next != nil {
sLog.next.Log(logType, level, logData, message) sLog.next.Log(logType, level, logData, message)

1
elektronLogging/types/types.go
View file

@ -1,6 +1,5 @@
package elektronLogging package elektronLogging
import "github.com/fatih/color"
const ( const (
ERROR = iota ERROR = iota

2
environment/env.go
View file

@ -26,3 +26,5 @@ var RaplPassword = "RAPL_PSSWD"
// Location of the script that sets the powercap value for a host. // Location of the script that sets the powercap value for a host.
var RaplThrottleScriptLocation = "RAPL_PKG_THROTTLE_SCRIPT_LOCATION" var RaplThrottleScriptLocation = "RAPL_PKG_THROTTLE_SCRIPT_LOCATION"
var LogConfigYaml = "logConfig.yaml"

29
go.mod
View file

@ -4,20 +4,23 @@ go 1.12
require ( require (
github.com/fatih/color v1.7.0 github.com/fatih/color v1.7.0
github.com/gogo/protobuf v1.1.1 github.com/gogo/protobuf v1.3.0
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
github.com/golang/protobuf v1.2.0 github.com/golang/protobuf v1.3.2
github.com/google/uuid v1.0.0 github.com/google/uuid v1.0.0
github.com/mash/gokmeans v0.0.0-20140614041449-8bbf08905a7e github.com/mash/gokmeans v0.0.0-20170215130432-ea22cff45f59
github.com/mattn/go-colorable v0.0.9 github.com/mattn/go-colorable v0.1.2
github.com/mattn/go-isatty v0.0.4 github.com/mattn/go-isatty v0.0.9
github.com/mesos/mesos-go v0.0.8 github.com/mesos/mesos-go v0.0.10
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe github.com/montanaflynn/stats v0.5.0
github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709 github.com/pborman/uuid v1.2.0
github.com/pkg/errors v0.8.0 github.com/pkg/errors v0.8.1
github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da
github.com/sirupsen/logrus v1.4.2
github.com/stretchr/testify v1.4.0 github.com/stretchr/testify v1.4.0
golang.org/x/crypto v0.0.0-20180927165925-5295e8364332 gitlab.com/spdf/elektron v0.0.0-20191024200717-26f96f361f10
golang.org/x/net v0.0.0-20180926154720-4dfa2610cdf3 golang.org/x/crypto v0.0.0-20190927123631-a832865fa7ad
golang.org/x/sys v0.0.0-20180928133829-e4b3c5e90611 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a
gopkg.in/yaml.v2 v2.2.2
) )

44
go.sum
View file

@ -1,41 +1,85 @@
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys= github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo= github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.3.0 h1:G8O7TerXerS4F6sx9OV7/nRfJdnXgHZu/S/7F2SN+UE=
github.com/gogo/protobuf v1.3.0/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA= github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA=
github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/mash/gokmeans v0.0.0-20140614041449-8bbf08905a7e h1:KrPVKjYg/ocMUbv39AQoIINny7acoDE0jghg+vTvTY4= github.com/mash/gokmeans v0.0.0-20140614041449-8bbf08905a7e h1:KrPVKjYg/ocMUbv39AQoIINny7acoDE0jghg+vTvTY4=
github.com/mash/gokmeans v0.0.0-20140614041449-8bbf08905a7e/go.mod h1:nxv+mR7KuDZRAmEjsCFBQNtWcIty99qUn2Ycbgcns5c= github.com/mash/gokmeans v0.0.0-20140614041449-8bbf08905a7e/go.mod h1:nxv+mR7KuDZRAmEjsCFBQNtWcIty99qUn2Ycbgcns5c=
github.com/mash/gokmeans v0.0.0-20170215130432-ea22cff45f59 h1:C7rqeTVPDC/0QZFxfsMvS2y66WjMxQogK8/PTy9Jo0A=
github.com/mash/gokmeans v0.0.0-20170215130432-ea22cff45f59/go.mod h1:nxv+mR7KuDZRAmEjsCFBQNtWcIty99qUn2Ycbgcns5c=
github.com/mattn/go-colorable v0.0.9 h1:UVL0vNpWh04HeJXV0KLcaT7r06gOH2l4OW6ddYRUIY4= github.com/mattn/go-colorable v0.0.9 h1:UVL0vNpWh04HeJXV0KLcaT7r06gOH2l4OW6ddYRUIY4=
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU=
github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
github.com/mattn/go-isatty v0.0.4 h1:bnP0vzxcAdeI1zdubAl5PjU6zsERjGZb7raWodagDYs= github.com/mattn/go-isatty v0.0.4 h1:bnP0vzxcAdeI1zdubAl5PjU6zsERjGZb7raWodagDYs=
github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
github.com/mattn/go-isatty v0.0.9 h1:d5US/mDsogSGW37IV293h//ZFaeajb69h+EHFsv2xGg=
github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
github.com/mesos/mesos-go v0.0.8 h1:hiAUHba+ycyZLxDiBUqKs91a0LbHZaAca988LzN19xM= github.com/mesos/mesos-go v0.0.8 h1:hiAUHba+ycyZLxDiBUqKs91a0LbHZaAca988LzN19xM=
github.com/mesos/mesos-go v0.0.8/go.mod h1:kPYCMQ9gsOXVAle1OsoY4I1+9kPu8GHkf88aV59fDr4= github.com/mesos/mesos-go v0.0.8/go.mod h1:kPYCMQ9gsOXVAle1OsoY4I1+9kPu8GHkf88aV59fDr4=
github.com/mesos/mesos-go v0.0.10 h1:+M/7Zlkvw4MolkLvXHfj6hkDsLLHOOU54CmOkOUaNBc=
github.com/mesos/mesos-go v0.0.10/go.mod h1:kPYCMQ9gsOXVAle1OsoY4I1+9kPu8GHkf88aV59fDr4=
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0= github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0=
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
github.com/montanaflynn/stats v0.5.0 h1:2EkzeTSqBB4V4bJwWrt5gIIrZmpJBcoIRGS2kWLgzmk=
github.com/montanaflynn/stats v0.5.0/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709 h1:zNBQb37RGLmJybyMcs983HfUfpkw9OTFD9tbBfAViHE= github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709 h1:zNBQb37RGLmJybyMcs983HfUfpkw9OTFD9tbBfAViHE=
github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709/go.mod h1:VyrYX9gd7irzKovcSS6BIIEwPRkP2Wm2m9ufcdFSJ34= github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709/go.mod h1:VyrYX9gd7irzKovcSS6BIIEwPRkP2Wm2m9ufcdFSJ34=
github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=
github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec h1:6ncX5ko6B9LntYM0YBRXkiSaZMmLYeZ/NWcmeB43mMY= github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec h1:6ncX5ko6B9LntYM0YBRXkiSaZMmLYeZ/NWcmeB43mMY=
github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E= github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da h1:p3Vo3i64TCLY7gIfzeQaUJ+kppEO5WQG3cL8iE8tGHU=
github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk= github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
gitlab.com/spdf/elektron v0.0.0-20191024200717-26f96f361f10 h1:8ySmVAfFtX2mY2o7vbJauJBflwlDk1+XA6Xlp2K6bRg=
gitlab.com/spdf/elektron v0.0.0-20191024200717-26f96f361f10/go.mod h1:F6knSIgL9ajNjb0oNAjzV0vpJ51FtNs1UHxw86QQAQ0=
golang.org/x/crypto v0.0.0-20180927165925-5295e8364332 h1:hvQVdF6P9DX4OiKA5tpehlG6JsgzmyQiThG7q5Bn3UQ= golang.org/x/crypto v0.0.0-20180927165925-5295e8364332 h1:hvQVdF6P9DX4OiKA5tpehlG6JsgzmyQiThG7q5Bn3UQ=
golang.org/x/crypto v0.0.0-20180927165925-5295e8364332/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20180927165925-5295e8364332/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190927123631-a832865fa7ad h1:5E5raQxcv+6CZ11RrBYQe5WRbUIWpScjh0kvHZkZIrQ=
golang.org/x/crypto v0.0.0-20190927123631-a832865fa7ad/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/net v0.0.0-20180926154720-4dfa2610cdf3 h1:dgd4x4kJt7G4k4m93AYLzM8Ni6h2qLTfh9n9vXJT3/0= golang.org/x/net v0.0.0-20180926154720-4dfa2610cdf3 h1:dgd4x4kJt7G4k4m93AYLzM8Ni6h2qLTfh9n9vXJT3/0=
golang.org/x/net v0.0.0-20180926154720-4dfa2610cdf3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180926154720-4dfa2610cdf3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/sys v0.0.0-20180928133829-e4b3c5e90611/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180928133829-e4b3c5e90611/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a h1:aYOabOQFp6Vj6W1F80affTUvO9UxmJRx8K0gsfABByQ=
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

28
pcp/pcp.go
View file

@ -20,16 +20,16 @@ package pcp
import ( import (
"bufio" "bufio"
"log"
"os/exec" "os/exec"
"syscall" "syscall"
"time" "time"
elekLogDef "github.com/spdfg/elektron/logging/def" "github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
func Start(quit chan struct{}, logging *bool, logMType chan elekLogDef.LogMessageType, func Start(quit chan struct{}, logging *bool,pcpConfigFile string) {
logMsg chan string, pcpConfigFile string) {
var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile
cmd := exec.Command("sh", "-c", pcpCommand) cmd := exec.Command("sh", "-c", pcpCommand)
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
@ -47,8 +47,9 @@ func Start(quit chan struct{}, logging *bool, logMType chan elekLogDef.LogMessag
scanner.Scan() scanner.Scan()
// Write to logfile // Write to logfile
logMType <- elekLogDef.PCP elektronLogging.ElektronLog.Log(elekLogT.PCP,
logMsg <- scanner.Text() log.InfoLevel,
log.Fields {}, scanner.Text())
// Throw away first set of results // Throw away first set of results
scanner.Scan() scanner.Scan()
@ -59,16 +60,18 @@ func Start(quit chan struct{}, logging *bool, logMType chan elekLogDef.LogMessag
text := scanner.Text() text := scanner.Text()
if *logging { if *logging {
logMType <- elekLogDef.PCP elektronLogging.ElektronLog.Log(elekLogT.PCP,
logMsg <- text log.InfoLevel,
log.Fields {}, text)
} }
seconds++ seconds++
} }
}(logging) }(logging)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "PCP logging started" log.InfoLevel,
log.Fields {}, "PCP logging started")
if err := cmd.Start(); err != nil { if err := cmd.Start(); err != nil {
log.Fatal(err) log.Fatal(err)
@ -78,8 +81,9 @@ func Start(quit chan struct{}, logging *bool, logMType chan elekLogDef.LogMessag
select { select {
case <-quit: case <-quit:
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Stopping PCP logging in 5 seconds" log.InfoLevel,
log.Fields {}, "Stopping PCP logging in 5 seconds")
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
// http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly // http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly

79
powerCap/extrema.go
View file

@ -22,7 +22,6 @@ import (
"bufio" "bufio"
"container/ring" "container/ring"
"fmt" "fmt"
"log"
"os/exec" "os/exec"
"sort" "sort"
"strconv" "strconv"
@ -30,20 +29,23 @@ import (
"syscall" "syscall"
"time" "time"
elekLogDef "github.com/spdfg/elektron/logging/def"
"github.com/spdfg/elektron/pcp" "github.com/spdfg/elektron/pcp"
"github.com/spdfg/elektron/rapl" "github.com/spdfg/elektron/rapl"
"github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThreshold, loThreshold float64, func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThreshold, loThreshold float64,pcpConfigFile string) {
logMType chan elekLogDef.LogMessageType, logMsg chan string, pcpConfigFile string) {
var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile
cmd := exec.Command("sh", "-c", pcpCommand, pcpConfigFile) cmd := exec.Command("sh", "-c", pcpCommand, pcpConfigFile)
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
if hiThreshold < loThreshold { if hiThreshold < loThreshold {
log.Println("High threshold is lower than low threshold!") elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {}, "High threshold is lower than low threshold!")
} }
pipe, err := cmd.StdoutPipe() pipe, err := cmd.StdoutPipe()
@ -59,8 +61,9 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
scanner.Scan() scanner.Scan()
// Write to logfile // Write to logfile
logMType <- elekLogDef.PCP elektronLogging.ElektronLog.Log(elekLogT.PCP,
logMsg <- scanner.Text() log.InfoLevel,
log.Fields {}, scanner.Text())
headers := strings.Split(scanner.Text(), ",") headers := strings.Split(scanner.Text(), ",")
@ -95,12 +98,17 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
for scanner.Scan() { for scanner.Scan() {
if *logging { if *logging {
logMType <- elekLogDef.GENERAL
logMsg <- "Logging PCP..." elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {}, "Logging PCP...")
text := scanner.Text() text := scanner.Text()
split := strings.Split(text, ",") split := strings.Split(text, ",")
logMType <- elekLogDef.PCP
logMsg <- text elektronLogging.ElektronLog.Log(elekLogT.PCP,
log.InfoLevel,
log.Fields {}, text)
totalPower := 0.0 totalPower := 0.0
for _, powerIndex := range powerIndexes { for _, powerIndex := range powerIndexes {
@ -111,8 +119,10 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
powerHistories[host].Value = power powerHistories[host].Value = power
powerHistories[host] = powerHistories[host].Next() powerHistories[host] = powerHistories[host].Next()
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Host: %s, Power: %f", indexToHost[powerIndex], (power * pcp.RAPLUnits)) log.InfoLevel,
log.Fields {"Host" : fmt.Sprintf("%s",indexToHost[powerIndex]), "Power" : fmt.Sprintf("%f",(power * pcp.RAPLUnits))},
"")
totalPower += power totalPower += power
} }
@ -123,12 +133,16 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
clusterMean := pcp.AverageClusterPowerHistory(clusterPowerHist) clusterMean := pcp.AverageClusterPowerHistory(clusterPowerHist)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Total power: %f, %d Sec Avg: %f", clusterPower, clusterPowerHist.Len(), clusterMean) log.InfoLevel,
log.Fields {"Total power" : fmt.Sprintf("%f %d",clusterPower,clusterPowerHist.Len()),
"Sec Avg" : fmt.Sprintf("%f",clusterMean)},
"")
if clusterMean > hiThreshold { if clusterMean > hiThreshold {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Need to cap a node" log.InfoLevel,
log.Fields {}, "Need to cap a node")
// Create statics for all victims and choose one to cap // Create statics for all victims and choose one to cap
victims := make([]pcp.Victim, 0, 8) victims := make([]pcp.Victim, 0, 8)
@ -149,11 +163,14 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
if !cappedHosts[victim.Host] { if !cappedHosts[victim.Host] {
cappedHosts[victim.Host] = true cappedHosts[victim.Host] = true
orderCapped = append(orderCapped, victim.Host) orderCapped = append(orderCapped, victim.Host)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Capping Victim %s Avg. Wattage: %f", victim.Host, victim.Watts*pcp.RAPLUnits) log.InfoLevel,
log.Fields {"Capping Victim" : fmt.Sprintf("%s",victim.Host),
"Avg. Wattage" : fmt.Sprintf("%f", victim.Watts*pcp.RAPLUnits)}, "")
if err := rapl.Cap(victim.Host, "rapl", 50); err != nil { if err := rapl.Cap(victim.Host, "rapl", 50); err != nil {
logMType <- elekLogDef.ERROR elektronLogging.ElektronLog.Log(elekLogT.ERROR,
logMsg <- "Error capping host" log.ErrorLevel,
log.Fields {}, "Error capping host")
} }
break // Only cap one machine at at time. break // Only cap one machine at at time.
} }
@ -167,11 +184,13 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
cappedHosts[host] = false cappedHosts[host] = false
// User RAPL package to send uncap. // User RAPL package to send uncap.
log.Printf("Uncapping host %s", host) log.Printf("Uncapping host %s", host)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Uncapped host %s", host) log.InfoLevel,
log.Fields {"Uncapped host" : host}, "")
if err := rapl.Cap(host, "rapl", 100); err != nil { if err := rapl.Cap(host, "rapl", 100); err != nil {
logMType <- elekLogDef.ERROR elektronLogging.ElektronLog.Log(elekLogT.ERROR,
logMsg <- "Error capping host" log.ErrorLevel,
log.Fields {}, "Error capping host")
} }
} }
} }
@ -181,8 +200,9 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
} }
}(logging, hiThreshold, loThreshold) }(logging, hiThreshold, loThreshold)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "PCP logging started" log.InfoLevel,
log.Fields {}, "PCP logging started")
if err := cmd.Start(); err != nil { if err := cmd.Start(); err != nil {
log.Fatal(err) log.Fatal(err)
@ -192,8 +212,9 @@ func StartPCPLogAndExtremaDynamicCap(quit chan struct{}, logging *bool, hiThresh
select { select {
case <-quit: case <-quit:
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Stopping PCP logging in 5 seconds" log.InfoLevel,
log.Fields {}, "Stopping PCP logging in 5 seconds")
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
// http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly // http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly

131
powerCap/progressiveExtrema.go
View file

@ -22,7 +22,6 @@ import (
"bufio" "bufio"
"container/ring" "container/ring"
"fmt" "fmt"
"log"
"math" "math"
"os/exec" "os/exec"
"sort" "sort"
@ -32,10 +31,12 @@ import (
"time" "time"
"github.com/spdfg/elektron/constants" "github.com/spdfg/elektron/constants"
elekLogDef "github.com/spdfg/elektron/logging/def"
"github.com/spdfg/elektron/pcp" "github.com/spdfg/elektron/pcp"
"github.com/spdfg/elektron/rapl" "github.com/spdfg/elektron/rapl"
"github.com/spdfg/elektron/utilities" "github.com/spdfg/elektron/utilities"
"github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
func round(num float64) int { func round(num float64) int {
@ -48,16 +49,16 @@ func getNextCapValue(curCapValue float64, precision int) float64 {
return float64(round(curCapValue*output)) / output return float64(round(curCapValue*output)) / output
} }
func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiThreshold, loThreshold float64, func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiThreshold, loThreshold float64,pcpConfigFile string) {
logMType chan elekLogDef.LogMessageType, logMsg chan string, pcpConfigFile string) {
var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile var pcpCommand string = "pmdumptext -m -l -f '' -t 1.0 -d , -c " + pcpConfigFile
cmd := exec.Command("sh", "-c", pcpCommand, pcpConfigFile) cmd := exec.Command("sh", "-c", pcpCommand, pcpConfigFile)
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
if hiThreshold < loThreshold { if hiThreshold < loThreshold {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "High threshold is lower than low threshold!" log.InfoLevel,
log.Fields {}, "High threshold is lower than low threshold!")
} }
pipe, err := cmd.StdoutPipe() pipe, err := cmd.StdoutPipe()
@ -73,8 +74,9 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
scanner.Scan() scanner.Scan()
// Write to logfile // Write to logfile
logMType <- elekLogDef.PCP elektronLogging.ElektronLog.Log(elekLogT.PCP,
logMsg <- scanner.Text() log.InfoLevel,
log.Fields {}, scanner.Text())
headers := strings.Split(scanner.Text(), ",") headers := strings.Split(scanner.Text(), ",")
@ -113,13 +115,15 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
for scanner.Scan() { for scanner.Scan() {
if *logging { if *logging {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Logging PCP..." log.InfoLevel,
log.Fields {}, "Logging PCP...")
split := strings.Split(scanner.Text(), ",") split := strings.Split(scanner.Text(), ",")
text := scanner.Text() text := scanner.Text()
logMType <- elekLogDef.PCP elektronLogging.ElektronLog.Log(elekLogT.PCP,
logMsg <- text log.InfoLevel,
log.Fields {}, text)
totalPower := 0.0 totalPower := 0.0
for _, powerIndex := range powerIndexes { for _, powerIndex := range powerIndexes {
@ -130,10 +134,10 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
powerHistories[host].Value = power powerHistories[host].Value = power
powerHistories[host] = powerHistories[host].Next() powerHistories[host] = powerHistories[host].Next()
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Host: %s, Power %f", log.InfoLevel,
indexToHost[powerIndex], (power * pcp.RAPLUnits)) log.Fields {"Host" : fmt.Sprintf("%s",indexToHost[powerIndex]), "Power" : fmt.Sprintf("%f",(power * pcp.RAPLUnits))},
"")
totalPower += power totalPower += power
} }
clusterPower := totalPower * pcp.RAPLUnits clusterPower := totalPower * pcp.RAPLUnits
@ -143,16 +147,24 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
clusterMean := pcp.AverageClusterPowerHistory(clusterPowerHist) clusterMean := pcp.AverageClusterPowerHistory(clusterPowerHist)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Total power: %f, %d Sec Avg: %f", clusterPower, clusterPowerHist.Len(), clusterMean) log.InfoLevel,
log.Fields {"Total power" : fmt.Sprintf("%f %d",clusterPower,clusterPowerHist.Len()),
"Sec Avg" : fmt.Sprintf("%f",clusterMean)},
"")
if clusterMean >= hiThreshold { if clusterMean >= hiThreshold {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Need to cap a node" log.InfoLevel,
logMType <- elekLogDef.GENERAL log.Fields {}, "Need to cap a node")
logMsg <- fmt.Sprintf("Cap values of capped victims: %v", cappedVictims)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Cap values of victims to uncap: %v", orderCappedVictims) log.InfoLevel,
log.Fields {"Cap values of capped victims" : fmt.Sprintf("%v",cappedVictims)}, "")
elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {"Cap values of victims to uncap" : fmt.Sprintf("%v",orderCappedVictims)}, "")
// Create statics for all victims and choose one to cap // Create statics for all victims and choose one to cap
victims := make([]pcp.Victim, 0, 8) victims := make([]pcp.Victim, 0, 8)
@ -179,11 +191,15 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
} }
// Need to cap this victim. // Need to cap this victim.
if err := rapl.Cap(victims[i].Host, "rapl", 50.0); err != nil { if err := rapl.Cap(victims[i].Host, "rapl", 50.0); err != nil {
logMType <- elekLogDef.GENERAL
logMsg <- fmt.Sprintf("Error capping host %s", victims[i].Host) elektronLogging.ElektronLog.Log(elekLogT.ERROR,
log.ErrorLevel,
log.Fields {"Error capping host" : fmt.Sprintf("%s",victims[i].Host)}, "")
} else { } else {
logMType <- elekLogDef.GENERAL
logMsg <- fmt.Sprintf("Capped host[%s] at %f", victims[i].Host, 50.0) elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {}, fmt.Sprintf("Capped host[%s] at %f", victims[i].Host, 50.0))
// Keeping track of this victim and it's cap value // Keeping track of this victim and it's cap value
cappedVictims[victims[i].Host] = 50.0 cappedVictims[victims[i].Host] = 50.0
newVictimFound = true newVictimFound = true
@ -206,12 +222,15 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
if capValue > constants.LowerCapLimit { if capValue > constants.LowerCapLimit {
newCapValue := getNextCapValue(capValue, 2) newCapValue := getNextCapValue(capValue, 2)
if err := rapl.Cap(alreadyCappedHosts[i], "rapl", newCapValue); err != nil { if err := rapl.Cap(alreadyCappedHosts[i], "rapl", newCapValue); err != nil {
logMType <- elekLogDef.ERROR
logMsg <- fmt.Sprintf("Error capping host[%s]", alreadyCappedHosts[i]) elektronLogging.ElektronLog.Log(elekLogT.ERROR,
log.ErrorLevel,
log.Fields {"Error capping host" : fmt.Sprintf("%s",alreadyCappedHosts[i])}, "")
} else { } else {
// Successful cap // Successful cap
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Capped host[%s] at %f", alreadyCappedHosts[i], newCapValue) log.InfoLevel,
log.Fields {}, fmt.Sprintf("Capped host[%s] at %f", alreadyCappedHosts[i], newCapValue))
// Checking whether this victim can be capped further // Checking whether this victim can be capped further
if newCapValue <= constants.LowerCapLimit { if newCapValue <= constants.LowerCapLimit {
// Deleting victim from cappedVictims. // Deleting victim from cappedVictims.
@ -234,18 +253,23 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
} }
} }
if !canCapAlreadyCappedVictim { if !canCapAlreadyCappedVictim {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "No Victim left to cap." log.InfoLevel,
log.Fields {}, "No Victim left to cap")
} }
} }
} else if clusterMean < loThreshold { } else if clusterMean < loThreshold {
logMType <- elekLogDef.GENERAL
logMsg <- "Need to uncap a node" elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMType <- elekLogDef.GENERAL log.InfoLevel,
logMsg <- fmt.Sprintf("Cap values of capped victims: %v", cappedVictims) log.Fields {}, "Need to uncap a node")
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Cap values of victims to uncap: %v", orderCappedVictims) log.InfoLevel,
log.Fields {"Cap values of capped victims" : fmt.Sprintf("%v",cappedVictims)}, "")
elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {"Cap values of victims to uncap" : fmt.Sprintf("%v",orderCappedVictims)}, "")
if len(orderCapped) > 0 { if len(orderCapped) > 0 {
// We pick the host that is capped the most to uncap. // We pick the host that is capped the most to uncap.
orderCappedToSort := utilities.GetPairList(orderCappedVictims) orderCappedToSort := utilities.GetPairList(orderCappedVictims)
@ -255,12 +279,15 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
// This is a floating point operation and might suffer from precision loss. // This is a floating point operation and might suffer from precision loss.
newUncapValue := orderCappedVictims[hostToUncap] * 2.0 newUncapValue := orderCappedVictims[hostToUncap] * 2.0
if err := rapl.Cap(hostToUncap, "rapl", newUncapValue); err != nil { if err := rapl.Cap(hostToUncap, "rapl", newUncapValue); err != nil {
logMType <- elekLogDef.ERROR
logMsg <- fmt.Sprintf("Error uncapping host[%s]", hostToUncap) elektronLogging.ElektronLog.Log(elekLogT.ERROR,
log.ErrorLevel,
log.Fields {"Error uncapping host" : fmt.Sprintf("%s",hostToUncap)}, "")
} else { } else {
// Successful uncap // Successful uncap
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- fmt.Sprintf("Uncapped host[%s] to %f", hostToUncap, newUncapValue) log.InfoLevel,
log.Fields {}, fmt.Sprintf("Uncapped host[%s] to %f", hostToUncap, newUncapValue))
// Can we uncap this host further. If not, then we remove its entry from orderCapped // Can we uncap this host further. If not, then we remove its entry from orderCapped
if newUncapValue >= 100.0 { // can compare using == if newUncapValue >= 100.0 { // can compare using ==
// Deleting entry from orderCapped // Deleting entry from orderCapped
@ -281,8 +308,9 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
} }
} }
} else { } else {
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "No host staged for Uncapped" log.InfoLevel,
log.Fields {}, "No host staged for Uncapped")
} }
} }
} }
@ -291,9 +319,9 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
}(logging, hiThreshold, loThreshold) }(logging, hiThreshold, loThreshold)
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "PCP logging started" log.InfoLevel,
log.Fields {}, "PCP logging started")
if err := cmd.Start(); err != nil { if err := cmd.Start(); err != nil {
log.Fatal(err) log.Fatal(err)
} }
@ -302,8 +330,9 @@ func StartPCPLogAndProgressiveExtremaCap(quit chan struct{}, logging *bool, hiTh
select { select {
case <-quit: case <-quit:
logMType <- elekLogDef.GENERAL elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
logMsg <- "Stopping PCP logging in 5 seconds" log.InfoLevel,
log.Fields {}, "Stopping PCP logging in 5 seconds")
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
// http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly // http://stackoverflow.com/questions/22470193/why-wont-go-kill-a-child-process-correctly

40
scheduler.go
View file

@ -21,20 +21,20 @@ package main // import github.com/spdfg/elektron
import ( import (
"flag" "flag"
"fmt" "fmt"
"log"
"os" "os"
"os/signal" "os/signal"
"strings"
"time" "time"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto" mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"
"github.com/spdfg/elektron/def" "github.com/spdfg/elektron/def"
elekLogDef "github.com/spdfg/elektron/logging/def"
"github.com/spdfg/elektron/pcp" "github.com/spdfg/elektron/pcp"
"github.com/spdfg/elektron/powerCap" "github.com/spdfg/elektron/powerCap"
"github.com/spdfg/elektron/schedulers" "github.com/spdfg/elektron/schedulers"
"github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
var master = flag.String("master", "", "Location of leading Mesos master -- <mesos-master>:<port>") var master = flag.String("master", "", "Location of leading Mesos master -- <mesos-master>:<port>")
@ -95,10 +95,6 @@ func main() {
os.Exit(1) os.Exit(1)
} }
// Logging channels.
logMType := make(chan elekLogDef.LogMessageType)
logMsg := make(chan string)
// First we need to build the scheduler using scheduler options. // First we need to build the scheduler using scheduler options.
var schedOptions []schedulers.SchedulerOptions = make([]schedulers.SchedulerOptions, 0, 10) var schedOptions []schedulers.SchedulerOptions = make([]schedulers.SchedulerOptions, 0, 10)
@ -123,7 +119,7 @@ func main() {
// Logging channels. // Logging channels.
// These channels are used by the framework to log messages. // These channels are used by the framework to log messages.
// The channels are used to send the type of log message and the message string. // The channels are used to send the type of log message and the message string.
schedOptions = append(schedOptions, schedulers.WithLoggingChannels(logMType, logMsg)) //schedOptions = append(schedOptions, schedulers.WithLoggingChannels(logMType, logMsg))
// Shutdown indicator channels. // Shutdown indicator channels.
// These channels are used to notify, // These channels are used to notify,
@ -228,28 +224,20 @@ func main() {
log.Fatal(fmt.Sprintf("Unable to create scheduler driver: %s", err)) log.Fatal(fmt.Sprintf("Unable to create scheduler driver: %s", err))
} }
// If here, then all command-line arguments validate.
// Creating logger and attaching different logging platforms.
startTime := time.Now()
formattedStartTime := startTime.Format("20060102150405")
// Checking if prefix contains any special characters. // Checking if prefix contains any special characters.
if strings.Contains(*pcplogPrefix, "/") { /*if strings.Contains(*pcplogPrefix, "/") {
log.Fatal("log file prefix should not contain '/'.") log.Fatal("log file prefix should not contain '/'.")
} }*/
logPrefix := *pcplogPrefix + "_" + formattedStartTime
logger := elekLogDef.BuildLogger(startTime, logPrefix)
// Starting the logging go-routine.
go logger.Listen(logMType, logMsg)
// Starting PCP logging. // Starting PCP logging.
if noPowercap { if noPowercap {
go pcp.Start(pcpLog, &recordPCP, logMType, logMsg, *pcpConfigFile) go pcp.Start(pcpLog, &recordPCP, *pcpConfigFile)
} else if extrema { } else if extrema {
go powerCap.StartPCPLogAndExtremaDynamicCap(pcpLog, &recordPCP, *hiThreshold, go powerCap.StartPCPLogAndExtremaDynamicCap(pcpLog, &recordPCP, *hiThreshold,
*loThreshold, logMType, logMsg, *pcpConfigFile) *loThreshold, *pcpConfigFile)
} else if progExtrema { } else if progExtrema {
go powerCap.StartPCPLogAndProgressiveExtremaCap(pcpLog, &recordPCP, *hiThreshold, go powerCap.StartPCPLogAndProgressiveExtremaCap(pcpLog, &recordPCP, *hiThreshold,
*loThreshold, logMType, logMsg, *pcpConfigFile) *loThreshold, *pcpConfigFile)
} }
// Take a second between starting PCP log and continuing. // Take a second between starting PCP log and continuing.
@ -284,8 +272,6 @@ func main() {
close(pcpLog) close(pcpLog)
time.Sleep(5 * time.Second) //Wait for PCP to log a few more seconds time.Sleep(5 * time.Second) //Wait for PCP to log a few more seconds
// Closing logging channels. // Closing logging channels.
close(logMType)
close(logMsg)
//case <-time.After(shutdownTimeout): //case <-time.After(shutdownTimeout):
} }
@ -296,7 +282,11 @@ func main() {
// Starting the scheduler driver. // Starting the scheduler driver.
if status, err := driver.Run(); err != nil { if status, err := driver.Run(); err != nil {
log.Printf("Framework stopped with status %s and error: %s\n", status.String(), err.Error()) elektronLogging.ElektronLog.Log(elekLogT.ERROR,
log.ErrorLevel,
log.Fields {"status" : status.String(), "error" : err.Error()}, "Framework stopped ")
} }
log.Println("Exiting...") elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {}, "Exiting...")
} }

2
schedulers/MaxGreedyMins.go
View file

@ -19,7 +19,7 @@
package schedulers package schedulers
import ( import (
"log" log "github.com/sirupsen/logrus"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto" mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"

2
schedulers/MaxMin.go
View file

@ -19,7 +19,7 @@
package schedulers package schedulers
import ( import (
"log" log "github.com/sirupsen/logrus"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto" mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"

199
schedulers/base.go
View file

@ -21,7 +21,6 @@ package schedulers
import ( import (
"bytes" "bytes"
"fmt" "fmt"
"log"
"sync" "sync"
"time" "time"
@ -30,9 +29,11 @@ import (
"github.com/mesos/mesos-go/api/v0/mesosutil" "github.com/mesos/mesos-go/api/v0/mesosutil"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"
"github.com/spdfg/elektron/def" "github.com/spdfg/elektron/def"
elekLogDef "github.com/spdfg/elektron/logging/def"
"github.com/spdfg/elektron/utilities" "github.com/spdfg/elektron/utilities"
"github.com/spdfg/elektron/utilities/schedUtils" "github.com/spdfg/elektron/utilities/schedUtils"
"github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
type BaseScheduler struct { type BaseScheduler struct {
@ -68,11 +69,6 @@ type BaseScheduler struct {
schedTrace *log.Logger schedTrace *log.Logger
// Send the type of the message to be logged
logMsgType chan elekLogDef.LogMessageType
// Send the message to be logged
logMsg chan string
mutex sync.Mutex mutex sync.Mutex
// Whether switching of scheduling policies at runtime has been enabled // Whether switching of scheduling policies at runtime has been enabled
@ -252,179 +248,170 @@ func (s *BaseScheduler) StatusUpdate(driver sched.SchedulerDriver, status *mesos
} }
} }
func (s *BaseScheduler) Log(lmt elekLogDef.LogMessageType, msg string) {
s.mutex.Lock()
s.logMsgType <- lmt
s.logMsg <- msg
s.mutex.Unlock()
}
func (s *BaseScheduler) LogTaskStarting(ts *def.Task, offer *mesos.Offer) { func (s *BaseScheduler) LogTaskStarting(ts *def.Task, offer *mesos.Offer) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt]
var msg string
if ts == nil { if ts == nil {
msg = msgColor.Sprintf("TASKS STARTING... host = [%s]", offer.GetHostname()) elektronLogging.ElektronLog.Log(lmt, log.InfoLevel,
log.Fields {"host" : fmt.Sprintf("%s",offer.GetHostname())}, "TASKS STARTING...")
} else { } else {
msg = msgColor.Sprintf("TASK STARTING... task = [%s], Instance = %d, host = [%s]", elektronLogging.ElektronLog.Log(lmt,
ts.Name, *ts.Instances, offer.GetHostname()) log.InfoLevel,
log.Fields {"task" : fmt.Sprintf("%s",ts.Name),
"Instance" : fmt.Sprintf("%d",*ts.Instances), "host" : fmt.Sprintf("%s",offer.GetHostname())},
"TASK STARTING... ")
} }
s.Log(lmt, msg)
} }
func (s *BaseScheduler) LogTaskWattsConsideration(ts def.Task, host string, wattsToConsider float64) { func (s *BaseScheduler) LogTaskWattsConsideration(ts def.Task, host string, wattsToConsider float64) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("Watts considered for task[%s] and host[%s] = %f Watts", log.InfoLevel,
ts.Name, host, wattsToConsider) log.Fields {"task" : ts.Name, "host" : host, "Watts" : fmt.Sprintf("%f",wattsToConsider)}, "Watts considered for ")
s.Log(lmt, msg)
} }
func (s *BaseScheduler) LogOffersReceived(offers []*mesos.Offer) { func (s *BaseScheduler) LogOffersReceived(offers []*mesos.Offer) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("Received %d resource offers", len(offers)) log.InfoLevel,
s.Log(lmt, msg) log.Fields {"Resource offers received" : fmt.Sprintf("%d",len(offers))}, "")
} }
func (s *BaseScheduler) LogNoPendingTasksDeclineOffers(offer *mesos.Offer) { func (s *BaseScheduler) LogNoPendingTasksDeclineOffers(offer *mesos.Offer) {
lmt := elekLogDef.WARNING lmt := elekLogT.WARNING
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("DECLINING OFFER for host[%s]... "+ log.WarnLevel,
"No tasks left to schedule", offer.GetHostname()) log.Fields {"DECLINING OFFER for host" : fmt.Sprintf("%s",offer.GetHostname())}, "No tasks left to schedule ")
s.Log(lmt, msg)
} }
func (s *BaseScheduler) LogNumberOfRunningTasks() { func (s *BaseScheduler) LogNumberOfRunningTasks() {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("Number of tasks still Running = %d", s.tasksRunning) log.InfoLevel,
s.Log(lmt, msg) log.Fields {"Number of tasks still Running" : fmt.Sprintf("%d",s.tasksRunning)}, "")
} }
func (s *BaseScheduler) LogCoLocatedTasks(slaveID string) { func (s *BaseScheduler) LogCoLocatedTasks(slaveID string) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt]
buffer := bytes.Buffer{} buffer := bytes.Buffer{}
buffer.WriteString(fmt.Sprintln("Colocated with:"))
s.TasksRunningMutex.Lock() s.TasksRunningMutex.Lock()
for taskName := range s.Running[slaveID] { for taskName := range s.Running[slaveID] {
buffer.WriteString(fmt.Sprintln(taskName)) buffer.WriteString(fmt.Sprintln(taskName))
} }
s.TasksRunningMutex.Unlock() s.TasksRunningMutex.Unlock()
msg := msgColor.Sprintf(buffer.String()) elektronLogging.ElektronLog.Log(lmt,
s.Log(lmt, msg) log.InfoLevel,
log.Fields {"Colocated with" : fmt.Sprintf("%s",buffer.String())}, "")
} }
func (s *BaseScheduler) LogSchedTrace(taskToSchedule *mesos.TaskInfo, offer *mesos.Offer) { func (s *BaseScheduler) LogSchedTrace(taskToSchedule *mesos.TaskInfo, offer *mesos.Offer) {
msg := fmt.Sprint(offer.GetHostname() + ":" + taskToSchedule.GetTaskId().GetValue()) elektronLogging.ElektronLog.Log(elekLogT.SCHED_TRACE,
s.Log(elekLogDef.SCHED_TRACE, msg) log.InfoLevel,
log.Fields {offer.GetHostname() : fmt.Sprintf("%s",taskToSchedule.GetTaskId().GetValue())}, "")
} }
func (s *BaseScheduler) LogTerminateScheduler() { func (s *BaseScheduler) LogTerminateScheduler() {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprint("Done scheduling all tasks!") log.InfoLevel,
s.Log(lmt, msg) log.Fields {}, "Done scheduling all tasks!")
} }
func (s *BaseScheduler) LogInsufficientResourcesDeclineOffer(offer *mesos.Offer, func (s *BaseScheduler) LogInsufficientResourcesDeclineOffer(offer *mesos.Offer,
offerResources ...interface{}) { offerResources ...interface{}) {
lmt := elekLogDef.WARNING lmt := elekLogT.WARNING
msgColor := elekLogDef.LogMessageColors[lmt]
buffer := bytes.Buffer{} buffer := bytes.Buffer{}
buffer.WriteString(fmt.Sprintln("DECLINING OFFER... Offer has insufficient resources to launch a task")) buffer.WriteString(fmt.Sprintf("<CPU: %f, RAM: %f, Watts: %f>", offerResources...))
buffer.WriteString(fmt.Sprintf("Offer Resources <CPU: %f, RAM: %f, Watts: %f>", offerResources...)) elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprint(buffer.String()) log.WarnLevel,
s.Log(lmt, msg) log.Fields {"Offer Resources" : fmt.Sprintf("%s",buffer.String())}, "DECLINING OFFER... Offer has insufficient resources to launch a task")
} }
func (s *BaseScheduler) LogOfferRescinded(offerID *mesos.OfferID) { func (s *BaseScheduler) LogOfferRescinded(offerID *mesos.OfferID) {
lmt := elekLogDef.ERROR lmt := elekLogT.ERROR
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("OFFER RESCINDED: OfferID = %s", offerID) log.ErrorLevel,
s.Log(lmt, msg) log.Fields {"OfferID" : fmt.Sprintf("%s",offerID)}, "OFFER RESCINDED")
} }
func (s *BaseScheduler) LogSlaveLost(slaveID *mesos.SlaveID) { func (s *BaseScheduler) LogSlaveLost(slaveID *mesos.SlaveID) {
lmt := elekLogDef.ERROR lmt := elekLogT.ERROR
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("SLAVE LOST: SlaveID = %s", slaveID) log.ErrorLevel,
s.Log(lmt, msg) log.Fields {"SlaveID" : fmt.Sprintf("%s",slaveID)}, "SLAVE LOST")
} }
func (s *BaseScheduler) LogExecutorLost(executorID *mesos.ExecutorID, slaveID *mesos.SlaveID) { func (s *BaseScheduler) LogExecutorLost(executorID *mesos.ExecutorID, slaveID *mesos.SlaveID) {
lmt := elekLogDef.ERROR lmt := elekLogT.ERROR
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("EXECUTOR LOST: ExecutorID = %s, SlaveID = %s", executorID, slaveID) log.ErrorLevel,
s.Log(lmt, msg) log.Fields {"ExecutorID" : fmt.Sprintf("%s",executorID), "SlaveID" : fmt.Sprintf("%s", slaveID)}, "EXECUTOR LOST")
} }
func (s *BaseScheduler) LogFrameworkMessage(executorID *mesos.ExecutorID, func (s *BaseScheduler) LogFrameworkMessage(executorID *mesos.ExecutorID,
slaveID *mesos.SlaveID, message string) { slaveID *mesos.SlaveID, message string) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("Received Framework message from executor [%s]: %s", executorID, message) log.InfoLevel,
s.Log(lmt, msg) log.Fields {"Received Framework message from executor" : executorID}, message)
} }
func (s *BaseScheduler) LogMesosError(err string) { func (s *BaseScheduler) LogMesosError(err string) {
lmt := elekLogDef.ERROR lmt := elekLogT.ERROR
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("MESOS ERROR: %s", err) log.ErrorLevel,
s.Log(lmt, msg) log.Fields {"MESOS ERROR" : fmt.Sprintf("%v", err)}, "")
} }
func (s *BaseScheduler) LogElectronError(err error) { func (s *BaseScheduler) LogElectronError(err error) {
lmt := elekLogDef.ERROR lmt := elekLogT.ERROR
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("ELECTRON ERROR: %v", err) log.ErrorLevel,
s.Log(lmt, msg) log.Fields {"ELECTRON ERROR" : fmt.Sprintf("%v",err)}, "")
} }
func (s *BaseScheduler) LogFrameworkRegistered(frameworkID *mesos.FrameworkID, func (s *BaseScheduler) LogFrameworkRegistered(frameworkID *mesos.FrameworkID,
masterInfo *mesos.MasterInfo) { masterInfo *mesos.MasterInfo) {
lmt := elekLogDef.SUCCESS lmt := elekLogT.SUCCESS
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("FRAMEWORK REGISTERED! frameworkID = %s, master = %s", log.InfoLevel,
frameworkID, masterInfo) log.Fields {"frameworkID" : fmt.Sprintf("%s",frameworkID), "master" : fmt.Sprintf("%s",masterInfo)}, "FRAMEWORK REGISTERED!")
s.Log(lmt, msg)
} }
func (s *BaseScheduler) LogFrameworkReregistered(masterInfo *mesos.MasterInfo) { func (s *BaseScheduler) LogFrameworkReregistered(masterInfo *mesos.MasterInfo) {
lmt := elekLogDef.GENERAL lmt := elekLogT.GENERAL
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprintf("Framework re-registered with master %s", masterInfo) log.InfoLevel,
s.Log(lmt, msg) log.Fields {"master" : fmt.Sprintf("%s",masterInfo)}, "Framework re-registered")
} }
func (s *BaseScheduler) LogDisconnected() { func (s *BaseScheduler) LogDisconnected() {
lmt := elekLogDef.WARNING lmt := elekLogT.WARNING
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := msgColor.Sprint("Framework disconnected with master") log.WarnLevel,
s.Log(lmt, msg) log.Fields {}, "Framework disconnected with master")
} }
func (s *BaseScheduler) LogTaskStatusUpdate(status *mesos.TaskStatus) { func (s *BaseScheduler) LogTaskStatusUpdate(status *mesos.TaskStatus) {
var lmt elekLogDef.LogMessageType lmt := elekLogT.GENERAL
switch *status.State { switch *status.State {
case mesos.TaskState_TASK_ERROR, mesos.TaskState_TASK_FAILED, case mesos.TaskState_TASK_ERROR, mesos.TaskState_TASK_FAILED,
mesos.TaskState_TASK_KILLED, mesos.TaskState_TASK_LOST: mesos.TaskState_TASK_KILLED, mesos.TaskState_TASK_LOST:
lmt = elekLogDef.ERROR lmt = elekLogT.ERROR
case mesos.TaskState_TASK_FINISHED: case mesos.TaskState_TASK_FINISHED:
lmt = elekLogDef.SUCCESS lmt = elekLogT.SUCCESS
default: default:
lmt = elekLogDef.GENERAL lmt = elekLogT.GENERAL
} }
msgColor := elekLogDef.LogMessageColors[lmt] elektronLogging.ElektronLog.Log(lmt,
msg := elekLogDef.LogMessageColors[elekLogDef.GENERAL].Sprintf("Task Status received for task [%s] --> %s", log.InfoLevel,
*status.TaskId.Value, msgColor.Sprint(NameFor(status.State))) log.Fields {"task" : fmt.Sprintf("%s",*status.TaskId.Value), "state" : NameFor(status.State)}, "Task Status received")
s.Log(lmt, msg)
} }
func (s *BaseScheduler) LogSchedPolicySwitch(name string, nextPolicy SchedPolicyState) { func (s *BaseScheduler) LogSchedPolicySwitch(name string, nextPolicy SchedPolicyState) {
logSPS := func() { logSPS := func() {
s.Log(elekLogDef.SPS, name) elektronLogging.ElektronLog.Log(elekLogT.SPS,
log.InfoLevel,
log.Fields {"Name" : name}, "")
} }
if s.hasReceivedResourceOffers && (s.curSchedPolicy != nextPolicy) { if s.hasReceivedResourceOffers && (s.curSchedPolicy != nextPolicy) {
logSPS() logSPS()
@ -433,10 +420,14 @@ func (s *BaseScheduler) LogSchedPolicySwitch(name string, nextPolicy SchedPolicy
} }
// Logging the size of the scheduling window and the scheduling policy // Logging the size of the scheduling window and the scheduling policy
// that is going to schedule the tasks in the scheduling window. // that is going to schedule the tasks in the scheduling window.
s.Log(elekLogDef.SCHED_WINDOW, fmt.Sprintf("%d %s", s.schedWindowSize, name)) elektronLogging.ElektronLog.Log(elekLogT.SCHED_WINDOW,
log.InfoLevel,
log.Fields {"Window size" : fmt.Sprintf("%d",s.schedWindowSize), "Name" : name}, "")
} }
func (s *BaseScheduler) LogClsfnAndTaskDistOverhead(overhead time.Duration) { func (s *BaseScheduler) LogClsfnAndTaskDistOverhead(overhead time.Duration) {
// Logging the overhead in microseconds. // Logging the overhead in microseconds.
s.Log(elekLogDef.CLSFN_TASKDIST_OVERHEAD, fmt.Sprintf("%f", float64(overhead.Nanoseconds())/1000.0)) elektronLogging.ElektronLog.Log(elekLogT.CLSFN_TASKDIST_OVERHEAD,
log.InfoLevel,
log.Fields {"Overhead in microseconds" : fmt.Sprintf("%f", float64(overhead.Nanoseconds())/1000.0)}, "")
} }

3
schedulers/electronScheduler.go
View file

@ -24,7 +24,6 @@ import (
mesos "github.com/mesos/mesos-go/api/v0/mesosproto" mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"
"github.com/spdfg/elektron/def" "github.com/spdfg/elektron/def"
elekLogDef "github.com/spdfg/elektron/logging/def"
) )
// Implements mesos scheduler. // Implements mesos scheduler.
@ -38,8 +37,6 @@ type ElectronScheduler interface {
// Each of these functions are supposed to call the Log(...) that sends the // Each of these functions are supposed to call the Log(...) that sends the
// log message type, and the log message to the corresponding channels. // log message type, and the log message to the corresponding channels.
// Pass the logMessageType and the logMessage to the loggers for logging.
Log(logMType elekLogDef.LogMessageType, logMsg string)
// To be called when about to launch a task. // To be called when about to launch a task.
// Log message indicating that a task is about to start executing. // Log message indicating that a task is about to start executing.
// Also, log the host on which the task is going to be launched. // Also, log the host on which the task is going to be launched.

17
schedulers/helpers.go
View file

@ -25,19 +25,24 @@ import (
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/spdfg/elektron/constants" "github.com/spdfg/elektron/constants"
"github.com/spdfg/elektron/def" "github.com/spdfg/elektron/def"
elekLogDef "github.com/spdfg/elektron/logging/def"
"github.com/spdfg/elektron/utilities" "github.com/spdfg/elektron/utilities"
"github.com/spdfg/elektron/utilities/mesosUtils" "github.com/spdfg/elektron/utilities/mesosUtils"
"log" log "github.com/sirupsen/logrus"
"github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
) )
func coLocated(tasks map[string]bool, s BaseScheduler) { func coLocated(tasks map[string]bool, s BaseScheduler) {
for task := range tasks { for task := range tasks {
s.Log(elekLogDef.GENERAL, task) elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {"Task" : task}, "")
} }
s.Log(elekLogDef.GENERAL, "---------------------") elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {}, "---------------------")
} }
// Get the powerClass of the given hostname. // Get the powerClass of the given hostname.
@ -129,13 +134,13 @@ func WithPCPLog(pcpLog chan struct{}) SchedulerOptions {
} }
} }
func WithLoggingChannels(lmt chan elekLogDef.LogMessageType, msg chan string) SchedulerOptions { /*func WithLoggingChannels(lmt chan elekLogDef.LogMessageType, msg chan string) SchedulerOptions {
return func(s ElectronScheduler) error { return func(s ElectronScheduler) error {
s.(*BaseScheduler).logMsgType = lmt s.(*BaseScheduler).logMsgType = lmt
s.(*BaseScheduler).logMsg = msg s.(*BaseScheduler).logMsg = msg
return nil return nil
} }
} }*/
func WithSchedPolSwitchEnabled(enableSchedPolicySwitch bool, switchingCriteria string) SchedulerOptions { func WithSchedPolSwitchEnabled(enableSchedPolicySwitch bool, switchingCriteria string) SchedulerOptions {
return func(s ElectronScheduler) error { return func(s ElectronScheduler) error {

9
schedulers/schedPolicy.go
View file

@ -20,13 +20,14 @@ package schedulers
import ( import (
"fmt" "fmt"
"log"
"time" "time"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto" mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler" sched "github.com/mesos/mesos-go/api/v0/scheduler"
"github.com/spdfg/elektron/def" "github.com/spdfg/elektron/def"
elekLogDef "github.com/spdfg/elektron/logging/def" "github.com/spdfg/elektron/elektronLogging"
elekLogT "github.com/spdfg/elektron/elektronLogging/types"
log "github.com/sirupsen/logrus"
) )
type SchedPolicyContext interface { type SchedPolicyContext interface {
@ -89,7 +90,9 @@ func switchTaskDistBased(baseSchedRef *BaseScheduler) string {
// Determine the distribution of tasks in the new scheduling window. // Determine the distribution of tasks in the new scheduling window.
taskDist, err := def.GetTaskDistributionInWindow(baseSchedRef.schedWindowSize, baseSchedRef.tasks) taskDist, err := def.GetTaskDistributionInWindow(baseSchedRef.schedWindowSize, baseSchedRef.tasks)
baseSchedRef.LogClsfnAndTaskDistOverhead(time.Now().Sub(startTime)) baseSchedRef.LogClsfnAndTaskDistOverhead(time.Now().Sub(startTime))
baseSchedRef.Log(elekLogDef.GENERAL, fmt.Sprintf("Switching... TaskDistribution[%f]", taskDist)) elektronLogging.ElektronLog.Log(elekLogT.GENERAL,
log.InfoLevel,
log.Fields {"Task Distribution" : fmt.Sprintf("%s",taskDist)}, "Switching... ")
if err != nil { if err != nil {
// All the tasks in the window were only classified into 1 cluster. // All the tasks in the window were only classified into 1 cluster.
// Max-Min and Max-GreedyMins would work the same way as Bin-Packing for this situation. // Max-Min and Max-GreedyMins would work the same way as Bin-Packing for this situation.