Enable paralel execution of checks

infrawatch · Jun 13, 2019 · f5ec38f · f5ec38f
1 parent 7692b11
commit f5ec38f
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 15 deletions.
diff --git a/config/config.go b/config/config.go
@@ -111,6 +111,7 @@ func GetAgentConfigMetadata() map[string][]Parameter {
 			Parameter{"keepalive_interval", "20", []Validator{IntValidatorFactory()}},
 			Parameter{"tmp_base_dir", "/var/tmp/collectd-sensubility-checks", []Validator{}},
 			Parameter{"shell_path", "/usr/bin/sh", []Validator{}},
+			Parameter{"worker_count", "2", []Validator{IntValidatorFactory()}},
 			Parameter{"checks", "{}", []Validator{}},
 		},
 		"amqp1": []Parameter{

diff --git a/main/main.go b/main/main.go
@@ -25,7 +25,6 @@ func main() {
 	if err != nil {
 		panic(err.Error())
 	}
-	fmt.Printf(cfg.Sections["sensu"].Options["checks"].GetString())
 	sensuConnector, err := sensu.NewConnector(cfg)
 	if err != nil {
 		fmt.Println(err.Error())
@@ -47,24 +46,29 @@ func main() {
 
 	requests := make(chan interface{})
 	results := make(chan interface{})
+	wait := make(chan bool)
 	defer close(results)
 
 	sensuConnector.Start(requests, results)
 	sensuScheduler.Start(requests)
-	// mani loop for executing commands
-
-	for {
-		req := <-requests
-		switch req := req.(type) {
-		case sensu.CheckRequest:
-			res, err := sensuExecutor.Execute(req)
-			if err != nil {
-				//TODO: log warning
+	// spawn worker goroutines
+	workers := cfg.Sections["sensu"].Options["worker_count"].GetInt()
+	for i := 0; i < workers; i++ {
+		go func() {
+			for {
+				req := <-requests
+				switch req := req.(type) {
+				case sensu.CheckRequest:
+					res, err := sensuExecutor.Execute(req)
+					if err != nil {
+						//TODO: log warning
+					}
+					results <- res
+				default:
+					//TODO: log warning
+				}
 			}
-			results <- res
-		default:
-			//TODO: log warning
-		}
+		}()
 	}
-	fmt.Printf("End")
+	<-wait
 }
diff --git a/tests/config_test.go b/tests/config_test.go
@@ -65,6 +65,7 @@ func TestConfigValues(t *testing.T) {
 	assert.Equal(t, "/var/tmp/test.log", conf.Sections["default"].Options["log_file"].GetString(), "Did not parse correctly")
 	assert.Equal(t, false, conf.Sections["default"].Options["allow_exec"].GetBool(), "Did not parse correctly")
 	assert.Equal(t, 666, conf.Sections["amqp1"].Options["port"].GetInt(), "Did not parse correctly")
+	os.Remove(file.Name())
 }
 
 func TestValidators(t *testing.T) {
@@ -109,4 +110,5 @@ func TestValidators(t *testing.T) {
 	if err == nil {
 		t.Errorf("Failed to report validation error in constructor.")
 	}
+	os.Remove(file.Name())
 }