Modified for Readis

Author: flexconstructor

Diff for: docker/container_monitor_setup.go

@@ -23,7 +23,7 @@ var (
 
 // Create new system monitor instance.
 var (
-	monitor *container_monitor.ContainerMonitor
+	listener *container_monitor.RedisListener
 )
 
 // Main function.
@@ -72,10 +72,10 @@ func main() {
 			break
 		}
 	}
-	monitor  = container_monitor.NewContainerMonitor(redis_url)
 
-	go monitor.Run()
-	defer monitor.Stop()
+	listener = container_monitor.NewRedisListener(redis_url,"",0)
+	go listener.Listen()
+	defer listener.Close()
 
 	err = daemon.ServeSignals()
 	if err != nil {
@@ -87,6 +87,6 @@ func main() {
 // Terminate daemon.
 func terminateHandler(sig os.Signal) error {
 	log.Println("terminating system monitor...")
-	monitor.Stop()
+	listener.Close()
 	return daemon.ErrStop
 }

Diff for: monitor.go

@@ -11,54 +11,37 @@ import (
 type ContainerMonitor struct {
 	close_channel chan bool          // Channel for close signal.
 	info_factory  *SystemInfoFactory // System info factory.
-	client        *redis.Client
+	testID string
 }
 
 // Returns new ContainerMonitor instance.
-func NewContainerMonitor(redis_url string) *ContainerMonitor {
+//
+// params: client *redis.Client   Instance of Redis client.
+//         test_id string         Test ID.
+func newContainerMonitor(client *redis.Client, test_id string) *ContainerMonitor {
 	return &ContainerMonitor{
 		close_channel: make(chan bool),
-		info_factory:  NewSystemInfoFactory(),
-		client: redis.NewClient(&redis.Options{
-			Addr:     redis_url,
-			Password: "",
-			DB:       0,
-		}),
+		info_factory:  NewSystemInfoFactory(client),
+		testID:test_id,
 	}
 }
 
 // Runs the container monitor.
 // Just starts listen of unix socket.
 func (m *ContainerMonitor) Run() {
-	pong, err := m.client.Ping().Result()
-	defer m.client.Close()
-	if err != nil {
-		log.Printf("redis error: %s", err.Error())
+	for{
+		select{
+		case <- time.After(time.Second *2):
+		m.info_factory.UpdateSystemInfo(m.testID)
+		case <- m.close_channel:
 		return
-	}
-	log.Printf("pong: %v", pong)
-	for {
-		select {
-		case <-time.After(2 * time.Second):
-			m.writeSystemInfo()
-		case <-m.close_channel:
-			return
 		}
 	}
 }
 
-// Writes system info to redis DB.
-func (m *ContainerMonitor) writeSystemInfo() {
-	info := m.info_factory.GetSystemInfo()
-	log.Printf(string(info))
-	err := m.client.Set("system:info", info, 0).Err()
-	if err != nil {
-		log.Printf("WRITE DATA ERROR: %s", err.Error())
-		m.Stop()
-	}
-}
 
 // Close redis connection.
 func (m *ContainerMonitor) Stop() {
+	log.Println("stop test: %s",m.testID)
 	m.close_channel <- true
 }

Diff for: monitor_client.go

@@ -12,7 +12,7 @@ type ProcessInfo struct {
 	Cwd           string                  // Process file path.
 	CreateTime    string                  // Process creation UNIX time.
 	MemoryInfo    *process.MemoryInfoStat // Process memory usage info.
-	MemoryPersent float32                 // Usage virtual memory in percents.
-	NumThreads    int32                   // Process threads count.
-	CPUPersent    float64                 // CPU usage in percents.
+	MemoryPercent float64                 // Usage virtual memory in percents.
+	NumThreads    int64                   // Process threads count.
+	CPUPercent    float64                 // CPU usage in percents.
 }

Diff for: process_info.go

@@ -21,8 +21,8 @@ func (b ByCPU) Less(i, j int) bool {
 	if b[i] == nil || b[j] == nil {
 		return false
 	}
-	if b[i].CPUPersent > 0 || b[j].CPUPersent > 0 {
-		return b[i].CPUPersent > b[j].CPUPersent
+	if b[i].CPUPercent > 0 || b[j].CPUPercent > 0 {
+		return b[i].CPUPercent > b[j].CPUPercent
 	}
-	return b[i].MemoryPersent > b[j].MemoryPersent
+	return b[i].MemoryPercent > b[j].MemoryPercent
 }

Diff for: process_sorter.go

@@ -0,0 +1,105 @@
+package container_monitor
+import (
+	"gopkg.in/redis.v4"
+	"log"
+)
+type RedisListener struct{
+	Client *redis.Client       // Redis client
+	monitor *ContainerMonitor  // Container monitor.
+}
+
+// Returns new instance of Redis listener.
+// props: r_url    string   Redis server URL.
+//        password string   Redis server password.
+//        db       int      Redis server Data Base ID.
+func NewRedisListener(r_url string, password string, db int)(*RedisListener){
+	return &RedisListener{
+		Client: redis.NewClient(&redis.Options{
+			Addr:     r_url,
+			Password: password,
+			DB:       db,
+		}),
+	}
+}
+
+// Listens Redis pub/sub channel.
+func (l *RedisListener)Listen(){
+	defer l.Client.Close()
+	pubsub,err:= l.Client.Subscribe(STRESS_TEST_CHANNEL)
+	if(err != nil){
+		log.Printf("CLIENT SUBSCRIBE ERROR: %s",err.Error())
+		return
+	}
+	defer pubsub.Unsubscribe(STRESS_TEST_CHANNEL)
+	for{
+		err:=l.ping()
+		if(err != nil){
+			log.Printf("CLIENT PING ERROR: %s",err.Error())
+			return
+		}
+		mess,err:= pubsub.ReceiveMessage()
+		if(err == nil){
+			l.readRedisMessage(mess)
+		}
+	}
+}
+
+// Calls redis pub/sub channel.
+func (l *RedisListener)Call(test_id string, command string){
+	mess:= newRedisMessage(command, test_id)
+	message_string,err:= marshalRedisMessage(mess)
+	if(err != nil){
+		log.Printf("can not create start redis message %s",err.Error())
+		return
+	}
+	l.Client.Publish(STRESS_TEST_CHANNEL,message_string)
+}
+
+// Closes listener
+func (l *RedisListener)Close(){
+	l.Client.Close()
+}
+
+// Reads Redis pub/sub messages.
+func(l *RedisListener)readRedisMessage(mess *redis.Message){
+	message,err:=unmarshalRedisMessage(mess.Payload)
+	if(err != nil){
+		log.Printf("Can not unmarshall redis message: %s",err.Error())
+		return
+	}
+	if(message.Command == START_COMMAND) {
+		l.startTest(message.TestID)
+	}else{
+		l.stopTest(message.TestID)
+	}
+}
+
+// Starts gathering information about the container system.
+func (l *RedisListener)startTest(test_id string){
+	if(l.monitor != nil){
+		log.Println("ERROR: last test not finiched!")
+		return
+	}
+	l.monitor = newContainerMonitor(l.Client,test_id)
+	go l.monitor.Run()
+}
+// Stops gathering information about the container system.
+func (l *RedisListener)stopTest(test_id string){
+	if(l.monitor == nil){
+		log.Println("ERROR: test not started!")
+		return
+	}
+	l.monitor.Stop()
+	l.monitor = nil
+}
+
+// Pings redis pub/sub channel.
+func (l *RedisListener)ping()(error){
+ err:=l.Client.Ping().Err()
+	if(err != nil){
+		return err
+	}
+	return nil
+}
+
+

Diff for: redis_listener.go

@@ -0,0 +1,54 @@
+package container_monitor
+import (
+	"encoding/json"
+	"log"
+)
+const (
+	START_COMMAND="start_test"               // Stress test started.
+	STOP_COMMAND="stop_test"                 // Stress test stopped.
+	STRESS_TEST_CHANNEL="stress_test_client" // Redis channel name.
+)
+
+// Redis pub/sub message
+type redisMessage struct{
+	Command string
+	TestID string
+}
+
+// Returns new instance of Redis pub/sub message.
+//
+// params: command string   Kind of command.
+//         test_id string   Stress test ID.
+func newRedisMessage(command string, test_id string)(*redisMessage){
+	return &redisMessage{
+		Command:command,
+		TestID: test_id,
+	}
+}
+
+// Encodes Redis pub/sub message instance to JSON string.
+//
+// param: message *redisMessage   Instance of Redis pub/sub message.
+// return JSON string or error instance.
+func marshalRedisMessage(message *redisMessage)(string, error){
+message_string,err:= json.Marshal(message)
+	if(err != nil){
+		log.Printf("Can not marshal redis message %s",err.Error())
+		return err.Error(),err
+	}
+	return string(message_string), nil
+}
+
+// Decodes Redis pub/sub message JSON string to instance of *redisMessage
+//
+// param: message_string string   JSON message string.
+// return: Instance of *redisMessage or error instance.
+func unmarshalRedisMessage(message_string string)(*redisMessage, error){
+	mess:= &redisMessage{}
+	err:= json.Unmarshal([]byte(message_string),mess)
+	if(err != nil){
+		log.Printf("Can not unmarshal redis message %s",err.Error())
+		return nil, err
+	}
+	return mess,nil
+}