now with a working subscription model for checks

Author: Jason Playne

run.sh

@@ -1,3 +1,6 @@
 #!/bin/sh
 export GOPATH=`pwd`
+
+# set DEBUG to a non empty value to display debug information
+export DEBUG=
 go run src/sensu-client.go --config-file=src/config/config.json

src/plugins/checks/external.go

@@ -1,6 +1,7 @@
 package checks
 
 import (
+	"fmt"
 	"os/exec"
 	"plugins"
 )
@@ -14,17 +15,20 @@ type ExternalCheck struct {
 func (ec *ExternalCheck) Init(config plugins.PluginConfig) (string, error) {
 	// make sure that the command exists?
 	ec.name = config.Name
-	em.command = config.Command
+	ec.command = config.Command
 	return ec.name, nil
 }
 
 func (ec *ExternalCheck) Gather(r *plugins.Result) error {
-	cmd := exec.Command(em.command)
+	fmt.Printf("About to start command\n")
+	cmd := exec.Command(ec.command)
 	err := cmd.Run()
 	ec.checkStatus = plugins.UNKNOWN
 
 	out, errOut := cmd.CombinedOutput()
-	r.SetOutput(string(out))
+	fmt.Println("Output BELOW")
+	fmt.Println(out)
+	r.Add(string(out))
 	if nil == errOut {
 		ec.checkStatus = plugins.OK
 	} else {

src/plugins/metrics/external.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"fmt"
 	"os/exec"
 	"plugins"
 )
@@ -18,12 +19,16 @@ func (em *ExternalMetric) Init(config plugins.PluginConfig) (string, error) {
 }
 
 func (em *ExternalMetric) Gather(r *plugins.Result) error {
+	fmt.Printf("About to start command\n")
 	cmd := exec.Command(em.command)
 	err := cmd.Run()
 
 	out, errOut := cmd.CombinedOutput()
+	fmt.Println("Output BELOW")
+	fmt.Println(out)
+
 	if nil == errOut {
-		r.SetOutput(string(bytes))
+		r.Add(em.name + " " + string(out))
 	}
 
 	return err

src/plugins/plugins.go

@@ -13,13 +13,13 @@ type SensuPluginInterface interface {
 }
 
 type PluginConfig struct {
-	Type       string
-	Name       string
-	Command    string
-	Args       []string
-	Handlers   []string
-	Standalone bool
-	Interval   time.Duration
+	Type       string        `json:"type"`
+	Name       string        `json:"name"`
+	Command    string        `json:"command"`
+	Args       []string      `json:"args"`
+	Handlers   []string      `json:"handlers"`
+	Standalone bool          `json:"standalone"`
+	Interval   time.Duration `json:"interval"`
 }
 
 type Status int // check status - not used for metrics

src/sensu-client.go

@@ -25,10 +25,12 @@ func main() {
 		os.Exit(1)
 	}
 
+	log.SetPrefix("Sensu: ")
+
 	processes := []sensu.Processor{
-		new(sensu.Keepalive),
+		//new(sensu.Keepalive),
 		new(sensu.Subscriber),
-		sensu.NewPluginProcessor(),
+		//sensu.NewPluginProcessor(),
 	}
 	c := sensu.NewClient(settings, processes)
 

src/sensu/check.go

@@ -5,10 +5,11 @@ import (
 )
 
 type Check struct {
-	Name            string
-	Command         string
+	Name            string `json:"name"`
+	Command         string `json:"command"`
 	Executed        int
 	Status          int
+	Issued          int `json:"issued"`
 	Output          string
 	Duration        float64
 	Timeout         int

src/sensu/keepalive.go

@@ -5,13 +5,15 @@ import (
 	"github.com/bitly/go-simplejson"
 	"github.com/streadway/amqp"
 	"log"
+	"os"
 	"time"
 )
 
 type Keepalive struct {
 	q      MessageQueuer
 	config *Config
 	close  chan bool
+	logger *log.Logger
 }
 
 const keepaliveInterval = 20 * time.Second
@@ -23,6 +25,7 @@ func (k *Keepalive) Init(q MessageQueuer, config *Config) error {
 	); err != nil {
 		return fmt.Errorf("Exchange Declare: %s", err)
 	}
+	k.logger = log.New(os.Stdout, "Keepalive: ", log.LstdFlags)
 
 	k.q = q
 	k.config = config
@@ -61,10 +64,10 @@ func (k *Keepalive) publish(payload amqp.Publishing) {
 		"",
 		payload,
 	); err != nil {
-		log.Printf("keepalive.publish: %v", err)
+		k.logger.Printf("keepalive.publish: %v", err)
 		return
 	}
-	log.Print("Keepalive published")
+	k.logger.Print("Keepalive published")
 }
 
 func createKeepalivePayload(clientConfig *simplejson.Json, timestamp time.Time) amqp.Publishing {

src/sensu/plugin_processor.go

@@ -3,6 +3,7 @@ package sensu
 import (
 	"fmt"
 	"log"
+	"os"
 	"plugins"
 	"plugins/checks"
 	"plugins/metrics"
@@ -18,6 +19,7 @@ type PluginProcessor struct {
 	jobsConfig map[string]plugins.PluginConfig
 	close      chan bool
 	results    chan *Result
+	logger     *log.Logger
 }
 
 // used to create a new processor instance.
@@ -26,6 +28,7 @@ func NewPluginProcessor() *PluginProcessor {
 	proc.jobs = make(map[string]plugins.SensuPluginInterface)
 	proc.jobsConfig = make(map[string]plugins.PluginConfig)
 	proc.results = make(chan *Result, 500) // queue of 500 buffered results
+	proc.logger = log.New(os.Stdout, "Plugin: ", log.LstdFlags)
 
 	return proc
 }
@@ -78,10 +81,10 @@ func newCheckConfig(json interface{}) plugins.PluginConfig {
 func (p *PluginProcessor) AddJob(job plugins.SensuPluginInterface, checkConfig plugins.PluginConfig) {
 	name, err := job.Init(checkConfig)
 	if nil != err {
-		log.Printf("Failed to initialise check: (%s) %s\n", name, err)
+		p.logger.Printf("Failed to initialise check: (%s) %s\n", name, err)
 		return
 	}
-	log.Printf("Scheduling job: %s (%s) every %d seconds", name, checkConfig.Command, checkConfig.Interval)
+	p.logger.Printf("Scheduling job: %s (%s) every %d seconds", name, checkConfig.Command, checkConfig.Interval)
 
 	p.jobs[name] = job
 	p.jobsConfig[name] = checkConfig
@@ -107,39 +110,12 @@ func (p *PluginProcessor) Init(q MessageQueuer, config *Config) error {
 	for check_type, checkConfigInterface := range checks_config {
 		checkConfig, ok := checkConfigInterface.(map[string]interface{})
 		if !ok {
-			log.Printf("Failed to parse config: %", check_type)
+			p.logger.Printf("Failed to parse config: %", check_type)
 			continue
 		}
 
 		config := newCheckConfig(checkConfig)
-
-		// see if we can handle this check using one of our build in ones
-		switch check_type {
-		case "cpu_metrics":
-			check = new(metrics.CpuStats)
-		case "display_metrics":
-			check = new(metrics.DisplayStats)
-		case "interface_metrics":
-			check = new(metrics.NetworkInterfaceStats)
-		case "load_metrics":
-			check = new(metrics.LoadStats)
-		case "memory_metrics":
-			check = new(metrics.MemoryStats)
-		case "uptime_metrics":
-			check = new(metrics.UptimeStats)
-		case "wireless-ap_metrics":
-			check = new(metrics.WirelessStats)
-		case "check_procs":
-			check = new(checks.ProcessCheck)
-		default:
-			if "metric" == config.Type {
-				// we have a metric!
-				check = new(metrics.ExternalMetric)
-			} else {
-				// we have a check!
-				check = new(checks.ExternalCheck)
-			}
-		}
+		check = getCheckHandler(check_type, config.Type)
 
 		config.Name = check_type
 		p.AddJob(check, config)
@@ -153,17 +129,18 @@ func (p *PluginProcessor) Init(q MessageQueuer, config *Config) error {
 func (p *PluginProcessor) Start() {
 	go p.publish()
 
+	clientConfig := p.config.Data().Get("client")
+
 	// start our result publisher thread
 	for job_name, job := range p.jobs {
 		go func(theJobName string, theJob plugins.SensuPluginInterface) {
 			config := p.jobsConfig[job_name]
 
-			log.Printf("Starting job: %s", theJobName)
 			reset := make(chan bool)
 
 			timer := time.AfterFunc(0, func() {
-				log.Printf("Gathering: %s", theJobName)
-				result := NewResult(p.config.Data().Get("client"), theJobName)
+				p.logger.Printf("Gathering: %s", theJobName)
+				result := NewResult(clientConfig, theJobName)
 				result.SetCommand(config.Command)
 
 				presult := new(plugins.Result)
@@ -174,7 +151,7 @@ func (p *PluginProcessor) Start() {
 
 				if nil != err {
 					// returned an error - we should stop this job from running
-					log.Printf("Failed to gather stat: %s. %v", theJobName, err)
+					p.logger.Printf("Failed to gather stat: %s. %v", theJobName, err)
 					reset <- false
 					return
 				}
@@ -216,11 +193,42 @@ func (p *PluginProcessor) publish() {
 		case result := <-p.results:
 			if result.HasOutput() {
 				if err := p.q.Publish(RESULTS_QUEUE, "", result.GetPayload()); err != nil {
-					log.Printf("Error Publishing Stats: %v. %v", err, result)
+					p.logger.Printf("Error Publishing Stats: %v. %v", err, result)
 				}
 			}
 		case <-p.close:
 			return
 		}
 	}
 }
+
+func getCheckHandler(check_type, config_type string) plugins.SensuPluginInterface {
+	var check plugins.SensuPluginInterface
+	switch check_type {
+	case "cpu_metrics", "cpu_metrics.rb":
+		check = new(metrics.CpuStats)
+	case "display_metrics":
+		check = new(metrics.DisplayStats)
+	case "interface_metrics":
+		check = new(metrics.NetworkInterfaceStats)
+	case "load_metrics":
+		check = new(metrics.LoadStats)
+	case "memory_metrics":
+		check = new(metrics.MemoryStats)
+	case "uptime_metrics":
+		check = new(metrics.UptimeStats)
+	case "wireless-ap_metrics":
+		check = new(metrics.WirelessStats)
+	case "check_procs":
+		check = new(checks.ProcessCheck)
+	default:
+		if "metric" == config_type {
+			// we have a metric!
+			check = new(metrics.ExternalMetric)
+		} else {
+			// we have a check!
+			check = new(checks.ExternalCheck)
+		}
+	}
+	return check
+}

src/sensu/plugin_results.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"os"
 	"strings"
 	"time"
 )
@@ -139,7 +140,9 @@ func (result *Result) toJson() []byte {
 	if nil != err {
 		log.Panic(err)
 	}
-	log.Printf(string(json)) // handy json debug printing
+	if "" != os.Getenv("DEBUG") {
+		log.Printf(string(json)) // handy json debug printing
+	}
 	return json
 }
 

src/sensu/subscriber.go

@@ -1,28 +1,39 @@
 package sensu
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/streadway/amqp"
 	"log"
+	"os"
+	"plugins"
+	"time"
 )
 
 type Subscriber struct {
 	deliveries <-chan amqp.Delivery
 	done       chan error
+	logger     *log.Logger
+	config     *Config
+	q          MessageQueuer
 }
 
 func (s *Subscriber) Init(q MessageQueuer, c *Config) error {
 
+	s.config = c
+	s.q = q
 	config_name, _ := c.Data().Get("client").Get("name").String()
 	config_ver, _ := c.Data().Get("client").Get("version").String()
 
-	queue_name := config_name + "-" + config_ver
-	log.Printf("Declaring Queue: %s", queue_name)
+	s.logger = log.New(os.Stdout, "Subscriptions: ", log.LstdFlags)
+
+	queue_name := fmt.Sprintf("%s-%s-%d", config_name, config_ver, time.Now().Unix())
+	s.logger.Printf("Declaring Queue: %s", queue_name)
 	queue, err := q.QueueDeclare(queue_name)
 	if err != nil {
 		return fmt.Errorf("Queue Declare: %s", err)
 	}
-	log.Printf("declared Queue")
+	s.logger.Printf("declared Queue")
 
 	var subscriptions []string
 	subscriptions, err = c.Data().GetPath("client", "subscriptions").StringArray()
@@ -31,20 +42,20 @@ func (s *Subscriber) Init(q MessageQueuer, c *Config) error {
 	}
 
 	for _, sub := range subscriptions {
-		log.Printf("declaring Exchange (%q)", sub)
+		s.logger.Printf("declaring Exchange (%q)", sub)
 		err = q.ExchangeDeclare(sub, "fanout")
 		if err != nil {
 			return fmt.Errorf("Exchange Declare: %s", err)
 		}
 
-		log.Printf("binding %s to Exchange %q", queue.Name, sub)
+		s.logger.Printf("Binding %s to Exchange %q", queue.Name, sub)
 		err = q.QueueBind(queue.Name, "", sub)
 		if err != nil {
 			return fmt.Errorf("Queue Bind: %s", err)
 		}
 	}
 
-	log.Printf("starting Consume")
+	s.logger.Printf("Starting Consume on queue: " + queue.Name)
 	s.deliveries, err = q.Consume(queue.Name, "")
 	if err != nil {
 		return fmt.Errorf("Queue Consume: %s", err)
@@ -55,7 +66,7 @@ func (s *Subscriber) Init(q MessageQueuer, c *Config) error {
 }
 
 func (s *Subscriber) Start() {
-	go handle(s.deliveries, s.done)
+	go s.handle(s.deliveries, s.done)
 
 	// for {
 	// 	select {
@@ -69,15 +80,55 @@ func (s *Subscriber) Stop() {
 	s.done <- nil
 }
 
-func handle(deliveries <-chan amqp.Delivery, done chan error) {
+func (s *Subscriber) handle(deliveries <-chan amqp.Delivery, done chan error) {
+	clientConfig := s.config.Data().Get("client")
 	for d := range deliveries {
-		log.Printf(
+		s.logger.Printf(
 			"got %dB delivery: [%v] %q",
 			len(d.Body),
 			d.DeliveryTag,
 			d.Body,
 		)
+
+		checkConfig := new(plugins.PluginConfig)
+		err := json.Unmarshal(d.Body, checkConfig)
+		if nil != err {
+			s.logger.Printf("Unable to decode message, skipping...")
+			d.Reject(false)
+			continue
+		}
+
+		//s.logger.Printf("Our check consists of: %+v", checkConfig)
+
+		theJob := getCheckHandler(checkConfig.Name, checkConfig.Type)
+
+		result := NewResult(clientConfig, checkConfig.Name)
+		result.SetCommand(checkConfig.Command)
+
+		presult := new(plugins.Result)
+
+		theJob.Init(*checkConfig)
+
+		err = theJob.Gather(presult)
+		result.SetOutput(presult.Output())
+		result.SetCheckStatus(theJob.GetStatus())
+
+		if nil != err {
+			// returned an error - we should stop this job from running
+			s.logger.Printf("Failed to gather stat: %s. %v", checkConfig.Name, err)
+			return
+		}
+
+		// and now send it back
+		if result.HasOutput() {
+			if err = s.q.Publish(RESULTS_QUEUE, "", result.GetPayload()); err != nil {
+				s.logger.Printf("Error Publishing Stats: %v. %v", err, result)
+			}
+		}
+
+		d.Ack(false)
+
 	}
-	log.Printf("handle: deliveries channel closed")
+	s.logger.Printf("handle: deliveries channel closed")
 	done <- nil
 }