can execute checks concurrently

4 files changed, 68 insertions, 27 deletions

diff --git a/config.go b/config.go
index 660b49d..f54fabc 100644
--- a/config.go
+++ b/config.go
@@ -9,12 +9,13 @@ import (
 )
 
 type config struct {
-	EmailTo       string
-	EmailFrom     string
-	SMTPServer    string `json:"omitempty"`
-	StateDir      string `json:"omitempty"`
-	CheckTimeoutS int
-	Checks        map[string]check
+	EmailTo          string
+	EmailFrom        string
+	SMTPServer       string `json:"omitempty"`
+	StateDir         string `json:"omitempty"`
+	CheckTimeoutS    int
+	CheckConcurrency int
+	Checks           map[string]check
 }
 
 func newConfig(configFile string) (config, error) {
diff --git a/gogios.json b/gogios.json
index a6cbace..12f13aa 100644
--- a/gogios.json
+++ b/gogios.json
@@ -2,6 +2,7 @@
   "EmailTo": "paul",
   "EmailFrom": "gogios@buetow.org",
   "CheckTimeoutS": 10,
+  "CheckConcurrency": 2,
   "Checks": {
     "foo.zone HTTP IPv4": {
       "Plugin": "/usr/local/libexec/nagios/check_http",
diff --git a/main.go b/main.go
index 5227a1b..6d661d7 100644
--- a/main.go
+++ b/main.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"log"
+	"sync"
 	"time"
 )
 
@@ -21,20 +23,50 @@ func main() {
 		notifyError(config, err)
 	}
 
-	for name, check := range config.Checks {
-		ctx, cancel := context.WithTimeout(context.Background(),
-			time.Duration(config.CheckTimeoutS)*time.Second)
-		defer cancel()
+	type entry struct {
+		name  string
+		check check
+	}
 
-		output, status := check.execute(ctx)
-		stateChanged := state.update(name, status)
+	limiterCh := make(chan struct{}, config.CheckConcurrency)
+	checkCh := make(chan entry)
 
-		if status != ok || stateChanged {
-			subject := fmt.Sprintf("GOGIOS %s: %s", codeToString(status), name)
-			notify(config, subject, output)
+	go func() {
+		for name, check := range config.Checks {
+			checkCh <- entry{name, check}
 		}
+		close(checkCh)
+	}()
+
+	var wg sync.WaitGroup
+	wg.Add(len(config.Checks))
+
+	for entry := range checkCh {
+		go func(name string, check check) {
+			limiterCh <- struct{}{}
+			defer func() {
+				<-limiterCh
+				wg.Done()
+			}()
+
+			ctx, cancel := context.WithTimeout(context.Background(),
+				time.Duration(config.CheckTimeoutS)*time.Second)
+			defer cancel()
+
+			output, status := check.execute(ctx)
+			stateChanged := state.update(name, status)
+
+			if status != ok || stateChanged {
+				subject := fmt.Sprintf("GOGIOS %s: %s", codeToString(status), name)
+				notify(config, subject, output)
+			}
+
+		}(entry.name, entry.check)
 	}
 
+	wg.Wait()
+	log.Println("All checks completed!")
+
 	if err := state.persist(); err != nil {
 		notifyError(config, err)
 	}
diff --git a/state.go b/state.go
index 58c209f..d2be1c0 100644
--- a/state.go
+++ b/state.go
@@ -6,56 +6,63 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"sync"
 )
 
 type state struct {
 	stateFile string
 	Checks    map[string]int
+	mutex     sync.Mutex
 }
 
 func newState(config config) (state, error) {
-	stateFile := fmt.Sprintf("%s/state.json", config.StateDir)
-	state := state{stateFile, make(map[string]int)}
+	s := state{
+		stateFile: fmt.Sprintf("%s/state.json", config.StateDir),
+		Checks:    make(map[string]int),
+	}
 
-	if _, err := os.Stat(stateFile); err != nil {
+	if _, err := os.Stat(s.stateFile); err != nil {
 		// OK, may be first run with no state yet.
-		return state, nil
+		return s, nil
 	}
 
-	file, err := os.Open(stateFile)
+	file, err := os.Open(s.stateFile)
 	if err != nil {
-		return state, err
+		return s, err
 	}
 	defer file.Close()
 
 	// Read the file content
 	bytes, err := ioutil.ReadAll(file)
 	if err != nil {
-		return state, err
+		return s, err
 	}
 
 	// Parse the JSON content
-	if err := json.Unmarshal(bytes, &state.Checks); err != nil {
-		return state, err
+	if err := json.Unmarshal(bytes, &s.Checks); err != nil {
+		return s, err
 	}
 
 	// Clean up obsolete state information
 	var obsolete []string
-	for name := range state.Checks {
+	for name := range s.Checks {
 		if _, ok := config.Checks[name]; !ok {
 			obsolete = append(obsolete, name)
 		}
 	}
 
 	for _, name := range obsolete {
-		delete(state.Checks, name)
+		delete(s.Checks, name)
 		log.Printf("State of %s is obsolete (removed)", name)
 	}
 
-	return state, nil
+	return s, nil
 }
 
 func (s state) update(name string, status int) bool {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
 	oldStatus, ok := s.Checks[name]
 	if !ok {
 		log.Printf("State of %s: %d (new)", name, status)