diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..342baf1 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,22 @@ +# Contributing to Recon + +Recon is under active development. So, before you start writing code, discuss it by opening an issue. + +### Issues + +Please file bugs in the issues section detailing the problem completely. + +### Pull Requests + +*Before* submitting a pull request, please make sure the following is done + +1. Fork the repo and create your branch from `master`. +2. If you've added code that should be tested, add tests! +3. If you've changed APIs, update the documentation. +4. Ensure the code comments are succinct enough. +5. Ensure you have followed the [`Code Review Comments`](https://github.com/golang/go/wiki/CodeReviewComments) +5. Run your code through [`goimports`](https://godoc.org/golang.org/x/tools/cmd/goimports), [`golint`](https://github.com/golang/lint) and [`vet`](https://tip.golang.org/cmd/vet/) + +### License + +By contributing to Recon, you agree that your contributions will be licensed under its BSD license. \ No newline at end of file diff --git a/README.md b/README.md index ef80038..bfedd4b 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,17 @@ # Recon -Recon is an Open-source IT infrastructue & services monitoring tool. Our goal is to make an open-source tool that does what SaaS tools do, is easy to use, and is intelligent in its configurations, analytics and alerting. +Recon is an open source IT infrastructure & service monitoring tool. -### Release +#### Goals +* Ease of setup +* Simple UI +* Intelligent configuration +* Smart analytics and alerting -We are still in the early stages of development & will release a production ready build by September 2015. Follow the instructions below to install & try Recon out. +#### Release -### Installation +We are still in the early stages of development and will release a production ready build by September 2015. Follow the instructions below to install & try Recon out. + +#### Installation If you have Go installed and workspace setup, @@ -13,15 +19,20 @@ If you have Go installed and workspace setup, go get github.com/codeignition/recon/... ``` -### License +#### Terminology + +[`recond`](https://github.com/codeignition/recon/tree/master/cmd/recond) is the daemon (agent) that runs on your target machine (server). -BSD 3-clause "New" or "Revised" license +[`marksman`](https://github.com/codeignition/marksman) is the master server that aggregates the metrics from agents and exposes a public HTTP API. -### Disclaimer +#### Disclaimer So far the project is tested only on Linux, specifically Ubuntu 14.04. -### About +#### Contributing + +Check out the code and jump to the Issues section to join a conversation or to start one. Also, please read the [CONTRIBUTING](https://github.com/codeignition/recon/blob/master/CONTRIBUTING.md) document. -Check out the code & jump to Issues section to join a conversation or to start one for feedback or questions on the code, features and anything else. +#### License +BSD 3-clause "New" or "Revised" license \ No newline at end of file diff --git a/types.go b/agent.go similarity index 59% rename from types.go rename to agent.go index 6c6f503..5744991 100644 --- a/types.go +++ b/agent.go @@ -7,13 +7,6 @@ package recon // Agent represents a recon daemon running on // a machine. type Agent struct { - UID string `json:"uid"` -} - -type Metric struct { - // UID of the Agent - AgentUID string `json:"agent_uid"` - - // Metric data - Data map[string]interface{} `json:"metrics"` + UID string `json:"uid"` + HostName string `json:"host_name"` } diff --git a/cmd/recond/accumulate.go b/cmd/recond/accumulate.go deleted file mode 100644 index 8618f7a..0000000 --- a/cmd/recond/accumulate.go +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2015 CodeIgnition. All rights reserved. -// Use of this source code is governed by a BSD -// license that can be found in the LICENSE file. - -package main - -import ( - "log" - "os/user" - "time" - - "github.com/codeignition/recon/metrics/netstat" - "github.com/codeignition/recon/metrics/ps" -) - -func copyMap(from, to map[string]interface{}) { - for k, v := range from { - to[k] = v - } -} - -// accumulateData accumulates data from all other packages. -func accumulateData() map[string]interface{} { - currentUser, err := user.Current() - if err != nil { - log.Println(err) - } - psdata, err := ps.CollectData() - if err != nil { - log.Println(err) - } - nsdata, err := netstat.CollectData() - if err != nil { - log.Println(err) - } - data := map[string]interface{}{ - "recon_time": time.Now(), - "current_user": currentUser.Username, // if more data is required, use currentUser instead of just the Username field - "process_statistics": psdata, - "network_statistics": nsdata, - } - return data -} diff --git a/cmd/recond/agent.go b/cmd/recond/agent.go index 0ac17a8..085ec61 100644 --- a/cmd/recond/agent.go +++ b/cmd/recond/agent.go @@ -8,8 +8,6 @@ import ( "bytes" "encoding/json" "errors" - "fmt" - "io/ioutil" "net/http" "net/url" @@ -50,45 +48,23 @@ func (a *Agent) register(addr string) error { if err := dec.Decode(&t); err != nil { return err } - nc, err := nats.Connect(t.NatsURL) - if err != nil { - return err - } - // TODO: Should we return the conn instead of using a global? - natsEncConn, err = nats.NewEncodedConn(nc, "json") - if err != nil { - return err - } - return nil -} -func (a *Agent) update(addr string) error { - var buf bytes.Buffer - - m := recon.Metric{ - AgentUID: a.UID, - Data: accumulateData(), - } - - if err := json.NewEncoder(&buf).Encode(&m); err != nil { - return err + // Override the obtained NATS address with the one obtained from the command line flag. + var naddr string + if *flagNATSAddr == "" { + naddr = t.NatsURL + } else { + naddr = *flagNATSAddr } - l, err := url.Parse(addr + metricsAPIPath) + nc, err := nats.Connect(naddr) if err != nil { return err } - resp, err := http.Post(l.String(), "application/json", &buf) + // TODO: Should we return the conn instead of using a global? + natsEncConn, err = nats.NewEncodedConn(nc, "json") if err != nil { return err } - if resp.StatusCode != http.StatusCreated { - defer resp.Body.Close() - b, err := ioutil.ReadAll(resp.Body) - if err != nil { - return err - } - return fmt.Errorf("response status code not %d; response body: %s\n", http.StatusCreated, b) - } return nil } diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 6c9fcb1..cebdc20 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -7,16 +7,20 @@ package config import ( + "bytes" "crypto/rand" "encoding/json" + "errors" "fmt" "io" "log" "os" "os/user" "path/filepath" + "sync" "github.com/codeignition/recon/internal/fileutil" + "github.com/codeignition/recon/policy" ) const configFileName = ".recond.json" @@ -35,7 +39,10 @@ func init() { // Config represents the configuration for the recond // running on a particular machine. type Config struct { - UID string `json:"uid"` // Unique Identifier to register with marksman + sync.Mutex + UID string // Unique Identifier to register with marksman + HostName string + PolicyConfig policy.Config } // Init initializes and returns a Config. i.e. if the config file doesn't exist, @@ -53,8 +60,13 @@ func Init() (*Config, error) { if err != nil { return nil, err } + h, err := os.Hostname() + if err != nil { + return nil, err + } c := &Config{ - UID: uid, + UID: uid, + HostName: h, } err = c.Save() return c, err @@ -62,6 +74,7 @@ func Init() (*Config, error) { // Save saves the config in a configuration file. // If it already exists, it removes it and writes it freshly. +// Make sure you lock and unlock the config while calling Save. func (c *Config) Save() error { if fileutil.Exists(configPath) { if err := os.Remove(configPath); err != nil { @@ -78,10 +91,13 @@ func (c *Config) Save() error { // Many filesystems do their real work (and thus their real failures) on close. // You can defer a file.Close for Read, but not for write. - enc := json.NewEncoder(f) - if err := enc.Encode(c); err != nil { + var out bytes.Buffer + b, err := json.MarshalIndent(c, "", " ") + if err != nil { return err } + out.Write(b) + out.WriteTo(f) if err := f.Close(); err != nil { return err @@ -90,6 +106,18 @@ func (c *Config) Save() error { return nil } +func (c *Config) AddPolicy(p policy.Policy) error { + defer c.Unlock() + c.Lock() + for _, k := range c.PolicyConfig { + if k.Name == p.Name { + return errors.New("policy with the given name already exists") + } + } + c.PolicyConfig = append(c.PolicyConfig, p) + return nil +} + // parseConfig reads from a io.Reader and // creates a Config struct accordingly. // It takes an io.Reader so that it is easier diff --git a/cmd/recond/config/config_test.go b/cmd/recond/config/config_test.go index 215ad9b..548e9dc 100644 --- a/cmd/recond/config/config_test.go +++ b/cmd/recond/config/config_test.go @@ -13,6 +13,7 @@ import ( "testing" "github.com/codeignition/recon/internal/fileutil" + "github.com/codeignition/recon/policy" ) func TestGenerateUID(t *testing.T) { @@ -31,7 +32,7 @@ func TestGenerateUID(t *testing.T) { func TestParseConfig(t *testing.T) { fakeUID := "23fcdd694986" - fakeContent := fmt.Sprintf(`{"uid":"%s"} + fakeContent := fmt.Sprintf(`{"UID":"%s"} `, fakeUID) r := strings.NewReader(fakeContent) c, err := parseConfig(r) @@ -49,7 +50,7 @@ func TestInitExisting(t *testing.T) { t.Error(err) } fakeUID := "13fcdf794886" - fakeContent := fmt.Sprintf(`{"uid":"%s"} + fakeContent := fmt.Sprintf(`{"UID":"%s"} `, fakeUID) _, err = f.WriteString(fakeContent) if err != nil { @@ -94,7 +95,7 @@ func TestSave(t *testing.T) { t.Error(err) } fakeUID := "13fcdf794886" - fakeContent := fmt.Sprintf(`{"uid":"%s"} + fakeContent := fmt.Sprintf(`{"UID":"%s"} `, fakeUID) _, err = f.WriteString(fakeContent) if err != nil { @@ -131,3 +132,19 @@ func TestSave(t *testing.T) { t.Error(err) } } + +func TestConcurrentAddPolicy(t *testing.T) { + c := &Config{} + p := policy.Policy{ + Name: "foo", + Type: "bar", + } + // this will find any race conditions, if the tests are run using -race flag + // we ignore errors as this test only deals with finding out race conditions + go func() { + c.AddPolicy(p) + }() + go func() { + c.AddPolicy(p) + }() +} diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 3581bc4..2a48ef8 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -5,32 +5,42 @@ package main import ( + "errors" "flag" - "fmt" "log" - "time" + "sync" + + "golang.org/x/net/context" "github.com/codeignition/recon/cmd/recond/config" + "github.com/codeignition/recon/policy" + _ "github.com/codeignition/recon/policy/handlers" "github.com/nats-io/nats" ) -const ( - metricsAPIPath = "/api/metrics" // metrics path in the marksman server - agentsAPIPath = "/api/agents" // agents path in the marksman server -) +const agentsAPIPath = "/api/agents" // agents path in the marksman server // natsEncConn is the opened with the URL obtained from marksman. // It is populated if the agent registers successfully. var natsEncConn *nats.EncodedConn -// updateInterval is time.Duration that specifies the interval -// between two consecutive updates. -const updateInterval = 5 * time.Second +// ctxCancelFunc stores the map of policy name to +// the context cancel function. +var ctxCancelFunc = struct { + sync.Mutex + m map[string]context.CancelFunc +}{ + m: make(map[string]context.CancelFunc), +} + +var ( + flagNATSAddr = flag.String("nats", "", "address of the nats server, use only if you want to override the URL obtained from marksman") + flagMarksmanAddr = flag.String("marksman", "http://localhost:3000", "address of the marksman server") +) func main() { log.SetPrefix("recond: ") - var marksmanAddr = flag.String("marksman", "http://localhost:3000", "address of the marksman server") flag.Parse() conf, err := config.Init() @@ -41,25 +51,98 @@ func main() { // agent represents a single agent on which the recond // is running. var agent = &Agent{ - UID: conf.UID, + UID: conf.UID, + HostName: conf.HostName, } - err = agent.register(*marksmanAddr) + err = agent.register(*flagMarksmanAddr) if err != nil { log.Fatalln(err) } defer natsEncConn.Close() - natsEncConn.Subscribe(agent.UID, func(s string) { - fmt.Printf("Received a message: %s\n", s) - }) + if err := addSystemDataPolicy(conf); err != nil { + log.Fatal(err) + } + + go runStoredPolicies(conf) + + natsEncConn.Subscribe(agent.UID+"_add_policy", AddPolicyHandler(conf)) + natsEncConn.Subscribe(agent.UID+"_delete_policy", DeletePolicyHandler(conf)) + natsEncConn.Subscribe(agent.UID+"_modify_policy", ModifyPolicyHandler(conf)) + + // this is just to block the main function from exiting + c := make(chan struct{}) + <-c +} + +func runStoredPolicies(c *config.Config) { + for _, p := range c.PolicyConfig { + log.Printf("adding the policy %s...", p.Name) + go func(p policy.Policy) { + ctx, cancel := context.WithCancel(context.Background()) + events, err := p.Execute(ctx) + if err != nil { + log.Print(err) // TODO: send to a nats errors channel + } + ctxCancelFunc.Lock() + ctxCancelFunc.m[p.Name] = cancel + ctxCancelFunc.Unlock() + + for e := range events { + natsEncConn.Publish("policy_events", e) + } + }(p) + } +} + +func addSystemDataPolicy(c *config.Config) error { + // if the policy already exists, return silently + for _, p := range c.PolicyConfig { + if p.Name == "default_system_data" { + return nil + } + } + + p := policy.Policy{ + Name: "default_system_data", + AgentUID: c.UID, + Type: "system_data", + M: map[string]string{ + "interval": "5s", + }, + } + if err := c.AddPolicy(p); err != nil { + return err + + } + if err := c.Save(); err != nil { + return err + } + return nil +} + +func deletePolicy(c *config.Config, policyName string) error { + defer ctxCancelFunc.Unlock() + ctxCancelFunc.Lock() - c := time.Tick(updateInterval) - for now := range c { - log.Println("Update sent at", now) - if err := agent.update(*marksmanAddr); err != nil { - log.Println(err) + if _, ok := ctxCancelFunc.m[policyName]; !ok { + return errors.New("policy not found") + } + log.Printf("deleting the policy %s...", policyName) + + delete(ctxCancelFunc.m, policyName) + defer c.Unlock() + c.Lock() + for i, q := range c.PolicyConfig { + if q.Name == policyName { + c.PolicyConfig = append(c.PolicyConfig[:i], c.PolicyConfig[i+1:]...) } } + + if err := c.Save(); err != nil { + return err + } + return nil } diff --git a/cmd/recond/subscriptions.go b/cmd/recond/subscriptions.go new file mode 100644 index 0000000..c899781 --- /dev/null +++ b/cmd/recond/subscriptions.go @@ -0,0 +1,98 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package main + +import ( + "log" + + "github.com/codeignition/recon/cmd/recond/config" + "github.com/codeignition/recon/policy" + "golang.org/x/net/context" +) + +func AddPolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Policy) { + return func(subj, reply string, p *policy.Policy) { + log.Printf("add_policy received: %s\n", p.Name) + if err := conf.AddPolicy(*p); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + if err := conf.Save(); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + ctx, cancel := context.WithCancel(context.Background()) + events, err := p.Execute(ctx) + if err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + ctxCancelFunc.Lock() + ctxCancelFunc.m[p.Name] = cancel + ctxCancelFunc.Unlock() + + natsEncConn.Publish(reply, "add_policy_ack") // acknowledge policy add + for e := range events { + natsEncConn.Publish("policy_events", e) + } + } +} + +func DeletePolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Policy) { + return func(subj, reply string, p *policy.Policy) { + log.Printf("delete_policy received: %s\n", p.Name) + ctxCancelFunc.Lock() + cancel := ctxCancelFunc.m[p.Name] + ctxCancelFunc.Unlock() + cancel() + if err := deletePolicy(conf, p.Name); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + natsEncConn.Publish(reply, "delete_policy_ack") // acknowledge policy delete + } +} + +func ModifyPolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Policy) { + return func(subj, reply string, p *policy.Policy) { + log.Printf("modify_policy received: %s\n", p.Name) + + // We receive the complete policy with the new values + // and delete the old policy and stop its execution. + // Then we add the new policy. + ctxCancelFunc.Lock() + cancel := ctxCancelFunc.m[p.Name] + ctxCancelFunc.Unlock() + cancel() + if err := deletePolicy(conf, p.Name); err != nil { + log.Print(err) + natsEncConn.Publish(reply, err.Error()) + return + } + log.Printf("adding the policy %s...", p.Name) + if err := conf.AddPolicy(*p); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + if err := conf.Save(); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + ctx, cancel := context.WithCancel(context.Background()) + events, err := p.Execute(ctx) + if err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + ctxCancelFunc.Lock() + ctxCancelFunc.m[p.Name] = cancel + ctxCancelFunc.Unlock() + + natsEncConn.Publish(reply, "modify_policy_ack") // acknowledge policy delete + for e := range events { + natsEncConn.Publish("policy_events", e) + } + } +} diff --git a/metrics/blockdevice/blockdevice_linux.go b/metrics/misc/blockdevice/blockdevice_linux.go similarity index 100% rename from metrics/blockdevice/blockdevice_linux.go rename to metrics/misc/blockdevice/blockdevice_linux.go diff --git a/metrics/counters/counters.go b/metrics/misc/counters/counters.go similarity index 100% rename from metrics/counters/counters.go rename to metrics/misc/counters/counters.go diff --git a/metrics/cpu/cpu_linux.go b/metrics/misc/cpu/cpu_linux.go similarity index 100% rename from metrics/cpu/cpu_linux.go rename to metrics/misc/cpu/cpu_linux.go diff --git a/metrics/etc/etc_linux.go b/metrics/misc/etc/etc_linux.go similarity index 100% rename from metrics/etc/etc_linux.go rename to metrics/misc/etc/etc_linux.go diff --git a/metrics/filesystem/filesystem.go b/metrics/misc/filesystem/filesystem.go similarity index 100% rename from metrics/filesystem/filesystem.go rename to metrics/misc/filesystem/filesystem.go diff --git a/metrics/initpackage/initpackage_linux.go b/metrics/misc/initpackage/initpackage_linux.go similarity index 100% rename from metrics/initpackage/initpackage_linux.go rename to metrics/misc/initpackage/initpackage_linux.go diff --git a/metrics/kernel/kernel_linux.go b/metrics/misc/kernel/kernel_linux.go similarity index 100% rename from metrics/kernel/kernel_linux.go rename to metrics/misc/kernel/kernel_linux.go diff --git a/metrics/languages/languages.go b/metrics/misc/languages/languages.go similarity index 100% rename from metrics/languages/languages.go rename to metrics/misc/languages/languages.go diff --git a/metrics/lsb/lsb_linux.go b/metrics/misc/lsb/lsb_linux.go similarity index 100% rename from metrics/lsb/lsb_linux.go rename to metrics/misc/lsb/lsb_linux.go diff --git a/metrics/memory/memory_linux.go b/metrics/misc/memory/memory_linux.go similarity index 100% rename from metrics/memory/memory_linux.go rename to metrics/misc/memory/memory_linux.go diff --git a/metrics/netstat/netstat.go b/metrics/misc/netstat/netstat.go similarity index 100% rename from metrics/netstat/netstat.go rename to metrics/misc/netstat/netstat.go diff --git a/metrics/network/network.go b/metrics/misc/network/network.go similarity index 100% rename from metrics/network/network.go rename to metrics/misc/network/network.go diff --git a/metrics/ps/ps.go b/metrics/misc/ps/ps.go similarity index 100% rename from metrics/ps/ps.go rename to metrics/misc/ps/ps.go diff --git a/metrics/uptime/uptime_linux.go b/metrics/misc/uptime/uptime_linux.go similarity index 100% rename from metrics/uptime/uptime_linux.go rename to metrics/misc/uptime/uptime_linux.go diff --git a/metrics/system/disk/disk.go b/metrics/system/disk/disk.go new file mode 100644 index 0000000..9caeade --- /dev/null +++ b/metrics/system/disk/disk.go @@ -0,0 +1,52 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +// +build linux + +// Package disk provides disk metrics. +package disk + +import ( + "os/exec" + "strings" +) + +type Data map[string]interface{} + +func CollectData() (Data, error) { + d := make(Data) + d["disk"] = make(Data) + disk := d["disk"].(Data) + if err := sizeData(disk); err != nil { + return d, err + } + return d, nil +} + +func sizeData(d Data) error { + out, err := exec.Command("df", "-P").Output() + if err != nil { + return err + } + + lines := strings.Split(string(out), "\n") + + // lines[0] contains the column headings + // Filesystem 1024-blocks Used Available Capacity Mounted on + for _, line := range lines[1:] { + a := strings.Fields(line) + if len(a) >= 6 { + if strings.HasPrefix(a[0], "/dev/") { + d[a[0]] = map[string]interface{}{ + "kb_size": a[1], + "kb_used": a[2], + "kb_available": a[3], + "percentage_used": a[4], + "mounted_on": a[5], + } + } + } + } + return nil +} diff --git a/metrics/system/system.go b/metrics/system/system.go new file mode 100644 index 0000000..51583a9 --- /dev/null +++ b/metrics/system/system.go @@ -0,0 +1,40 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +// +build linux + +// Package system provides system metrics. +package system + +import ( + "github.com/codeignition/recon/metrics/system/disk" + "github.com/codeignition/recon/metrics/system/top" +) + +// Data denotes system data +type Data map[string]interface{} + +// Merge merges the input data with caller. +// It will overwrite the value of a key if it exists already. +func (d Data) Merge(a map[string]interface{}) { + for k, v := range a { + d[k] = v + } +} + +// CollectData collects the data and returns an error if any. +func CollectData() (Data, error) { + d := make(Data) + top, err := top.CollectData() + if err != nil { + return d, err + } + d.Merge(top) + disk, err := disk.CollectData() + if err != nil { + return d, err + } + d.Merge(disk) + return d, nil +} diff --git a/metrics/system/top/top.go b/metrics/system/top/top.go new file mode 100644 index 0000000..b4ef26c --- /dev/null +++ b/metrics/system/top/top.go @@ -0,0 +1,169 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +// +build linux + +// Package top provides selective data provided by the `top` command. +// It collects system summary data, current running tasks, etc. +package top + +import ( + "errors" + "fmt" + "os/exec" + "strconv" + "strings" +) + +type Data map[string]interface{} + +// CollectData collects the data and returns +// an error if any. +func CollectData() (Data, error) { + d := make(Data) + count := 0 // count of the top output iteration + iters := 2 // number of iterations that top command should collect + out, err := exec.Command("top", "-bn", strconv.Itoa(iters)).Output() + if err != nil { + return d, err + } + lines := strings.Split(string(out), "\n") // use a bufio.Scanner if memory problems arise + if len(lines) < 7 { + return d, errors.New("top: unexpected output") + } + base := 0 + for i, line := range lines { + if strings.HasPrefix(line, "top") { + count++ + base = i + } + if count == iters { + switch i - base { + case 0: + err := d.parseUptimeLoadAvgData(line) + if err != nil { + return d, err + } + case 1: + // we are not collecting the tasks data for now + continue + case 2: + err := d.parseCPUData(line) + if err != nil { + return d, err + } + case 3: + err := d.parseMemoryData(line) + if err != nil { + return d, err + } + case 4: + err := d.parseSwapData(line) + if err != nil { + return d, err + } + default: + break + } + } + } + return d, nil +} + +func (d Data) parseUptimeLoadAvgData(s string) error { + a := strings.SplitN(s, "load average: ", 2) + b := strings.SplitN(a[0], "up", 2) + t := strings.SplitN(b[1], "users", 2) + c := strings.TrimSpace(t[0]) + i := strings.LastIndex(c, ",") + if i == -1 { + return errors.New("top: unable to parse uptime data") + } + d["uptime"] = c[:i] + l := strings.Split(a[1], ",") + if len(l) != 3 { + return errors.New("top: unexpected number of load averages found") + } + var f [3]float64 + for i := range l { + x, err := strconv.ParseFloat(strings.TrimSpace(l[i]), 64) + f[i] = x + if err != nil { + return err + } + } + d["load_average"] = Data{ + "last_1_min": f[0], + "last_5_min": f[1], + "last_15_min": f[2], + } + return nil +} + +func (d Data) parseCPUData(s string) error { + a := strings.TrimPrefix(s, "%Cpu(s): ") + a = strings.TrimPrefix(a, "Cpu(s): ") // some systems have the prefix Cpu(s): without the % sign + b := strings.Split(a, ",") + if len(b) != 8 { + return errors.New("top: unknown number of CPU data") + } + var c [8]float64 + for i := range b { + t := strings.TrimSpace(b[i]) + x, err := strconv.ParseFloat(t[:len(t)-3], 64) + c[i] = x + if err != nil { + return err + } + } + d["cpu"] = Data{ + "userspace": c[0], + "idle": c[3], + "system": c[1], + "iowait": c[4], + "stolen": c[7], + } + return nil +} + +func (d Data) parseMemoryData(s string) error { + var total, used, free, buffers int + var err error + if strings.HasPrefix(s, "KiB Mem:") { + _, err = fmt.Sscanf(s, "KiB Mem:\t%d total,\t%d used,\t%d free,\t%d buffers", &total, &used, &free, &buffers) + } else if strings.HasPrefix(s, "Mem:") { + _, err = fmt.Sscanf(s, "Mem:\t%dk total,\t%dk used,\t%dk free,\t%dk buffers", &total, &used, &free, &buffers) + } + if err != nil { + return fmt.Errorf("top: unable to parse memory data; %s", err) + } + d["memory"] = Data{ + "total": total, + "used": used, + "free": free, + "buffers": buffers, + } + return nil +} + +func (d Data) parseSwapData(s string) error { + var total, used, free, cached int + var err error + if strings.HasPrefix(s, "KiB Swap:") { + _, err = fmt.Sscanf(s, "KiB Swap:\t%d total,\t%d used,\t%d free.\t%d cached Mem", &total, &used, &free, &cached) + } else if strings.HasPrefix(s, "Swap:") { + _, err = fmt.Sscanf(s, "Swap:\t%dk total,\t%dk used,\t%dk free,\t%dk cached", &total, &used, &free, &cached) + } + if err != nil { + return fmt.Errorf("top: unable to parse swap data; %s", err) + } + d["swap"] = Data{ + "total": total, + "used": used, + "free": free, + } + mem := d["memory"].(Data) + mem["cached"] = cached + return nil +} diff --git a/policy/event.go b/policy/event.go new file mode 100644 index 0000000..e4ee246 --- /dev/null +++ b/policy/event.go @@ -0,0 +1,15 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package policy + +import "time" + +// Event data that will be sent by policy handlers +type Event struct { + Time time.Time + PolicyName string `bson:"policy_name"` + AgentUID string `bson:"agent_uid"` + Data interface{} // Data may include status, stats, etc. +} diff --git a/policy/handlers/handlers.go b/policy/handlers/handlers.go new file mode 100644 index 0000000..1a7f757 --- /dev/null +++ b/policy/handlers/handlers.go @@ -0,0 +1,12 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package handlers + +import "github.com/codeignition/recon/policy" + +func init() { + policy.RegisterHandler("tcp", TCP) + policy.RegisterHandler("system_data", SystemData) +} diff --git a/policy/handlers/systemdata.go b/policy/handlers/systemdata.go new file mode 100644 index 0000000..90bd0da --- /dev/null +++ b/policy/handlers/systemdata.go @@ -0,0 +1,62 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package handlers + +import ( + "errors" + "log" + "time" + + "github.com/codeignition/recon/metrics/system" + "github.com/codeignition/recon/policy" + "golang.org/x/net/context" +) + +func SystemData(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { + interval, ok := p.M["interval"] + if !ok { + return nil, errors.New(`"interval" key missing in systemdata policy`) + } + d, err := time.ParseDuration(interval) + if err != nil { + return nil, err + } + // This check is here to ensure time.Ticker(d) doesn't panic + if d <= 0 { + return nil, errors.New("interval must be a positive quantity") + } + + out := make(chan policy.Event) + go func() { + t := time.NewTicker(d) + for { + select { + case <-ctx.Done(): + t.Stop() + close(out) + return + case <-t.C: + out <- policy.Event{ + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, + Data: accumulateSystemData(), + } + } + } + }() + return out, nil +} + +func accumulateSystemData() interface{} { + d, err := system.CollectData() + if err != nil { + log.Print(err) + } + a := map[string]interface{}{ + "system": d, + } + return a +} diff --git a/policy/handlers/tcp.go b/policy/handlers/tcp.go new file mode 100644 index 0000000..3974abb --- /dev/null +++ b/policy/handlers/tcp.go @@ -0,0 +1,80 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package handlers + +import ( + "errors" + "net" + "time" + + "github.com/codeignition/recon/policy" + + "golang.org/x/net/context" +) + +func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { + // Always use v, ok := p[key] form to avoid panic + addr, ok := p.M["address"] + if !ok { + return nil, errors.New(`"address" key missing in tcp policy`) + } + interval, ok := p.M["interval"] + if !ok { + return nil, errors.New(`"interval" key missing in tcp policy`) + } + + // From the time package docs: + // + // ParseDuration parses a duration string. + // A duration string is a possibly signed sequence of + // decimal numbers, each with optional fraction and a unit suffix, + // such as "300ms", "-1.5h" or "2h45m". + // Valid time units are "ns", "us", "ms", "s", "m", "h". + d, err := time.ParseDuration(interval) + if err != nil { + return nil, err + } + + // This check is here to ensure time.Ticker(d) doesn't panic + if d <= 0 { + return nil, errors.New("interval must be a positive quantity") + } + + out := make(chan policy.Event) + go func() { + t := time.NewTicker(d) + for { + select { + case <-ctx.Done(): + t.Stop() + close(out) + return + case <-t.C: + _, err := net.DialTimeout("tcp", addr, d) + if err != nil { + out <- policy.Event{ + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, + Data: map[string]interface{}{ + "status": "failure", + "error": err.Error(), + }, + } + } else { + out <- policy.Event{ + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, + Data: map[string]interface{}{ + "status": "success", + }, + } + } + } + } + }() + return out, nil +} diff --git a/policy/policy.go b/policy/policy.go new file mode 100644 index 0000000..c9c7068 --- /dev/null +++ b/policy/policy.go @@ -0,0 +1,79 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package policy + +import ( + "errors" + "sync" + + "golang.org/x/net/context" +) + +// Policy represents a monitoring policy +type Policy struct { + Name string // Name of the monitoring policy + AgentUID string // Agent UID + Type string // Type denotes the monitoring policy type. e.g. "tcp" + M map[string]string // M is the map containing the rules of a particular monitoring policy. +} + +// Config is a slice of monitoring policies +type Config []Policy + +// HandlerFunc is the type of a policy handler function. Any policy handler +// function must be of this type. +type HandlerFunc func(context.Context, Policy) (<-chan Event, error) + +// handlerFuncMap maps a policy type to a handler function +var handlerFuncMap = struct { + sync.Mutex + m map[string]HandlerFunc +}{ + m: make(map[string]HandlerFunc), +} + +func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { + if err := p.Valid(); err != nil { + return nil, err + } + + handlerFuncMap.Lock() + f := handlerFuncMap.m[p.Type] + handlerFuncMap.Unlock() + + return f(ctx, p) +} + +// Valid checks whether the policy is valid. +func (p Policy) Valid() error { + if p.Name == "" { + return errors.New("policy name can't be empty") + } + + handlerFuncMap.Lock() + _, ok := handlerFuncMap.m[p.Type] + handlerFuncMap.Unlock() + + if !ok { + return errors.New("policy type unknown") + } + return nil +} + +func RegisterHandler(policyType string, handlerFunc HandlerFunc) error { + if policyType == "" { + return errors.New("policy type can't be empty") + } + + handlerFuncMap.Lock() + defer handlerFuncMap.Unlock() + + if _, ok := handlerFuncMap.m[policyType]; ok { + return errors.New("handler for the policy type already exists") + } + + handlerFuncMap.m[policyType] = handlerFunc + return nil +} diff --git a/policy/policy_test.go b/policy/policy_test.go new file mode 100644 index 0000000..7ac5411 --- /dev/null +++ b/policy/policy_test.go @@ -0,0 +1,152 @@ +// Copyright 2015 CodeIgnition. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package policy + +import ( + "errors" + "testing" + "time" + + "golang.org/x/net/context" +) + +func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { + foo, ok := p.M["foo"] + if !ok { + return nil, errors.New(`"foo" key missing in fake policy`) + } + interval, ok := p.M["interval"] + if !ok { + return nil, errors.New(`"interval" key missing in fake policy`) + } + d, err := time.ParseDuration(interval) + if err != nil { + return nil, err + } + + // This check is here to ensure time.Ticker(d) doesn't panic + if d <= 0 { + return nil, errors.New("interval must be a positive quantity") + } + + out := make(chan Event) + go func() { + t := time.NewTicker(d) + for { + select { + case <-ctx.Done(): + t.Stop() + close(out) + return + case <-t.C: + out <- Event{ + Time: time.Now(), + Policy: p, + Data: map[string]interface{}{ + "foo": foo, + }, + } + } + } + }() + return out, nil +} + +func TestRegisterHandler(t *testing.T) { + err := RegisterHandler("", fakePolicyHandler) + if err == nil { + t.Fatal(`NewHandler should return an error when the type is empty`) + } + err = RegisterHandler("fake", fakePolicyHandler) + if err != nil { + t.Fatal(err) + } + + // test registering twice + err = RegisterHandler("fake", fakePolicyHandler) + if err == nil { + t.Fatal(`want error "handler for the policy type already exists"; got nil`) + } +} + +func TestRegisterHandlerConcurrent(t *testing.T) { + f1 := new(HandlerFunc) + f2 := new(HandlerFunc) + // This checks for a data race when `go test -race` is executed + go func() { + err := RegisterHandler("foo", *f1) + if err != nil { + t.Error(err) + } + }() + err := RegisterHandler("foo", *f2) + if err != nil { + t.Error(err) + } +} + +func TestValid(t *testing.T) { + p := Policy{ + Name: "", + } + if err := p.Valid(); err == nil { + t.Error(`want error "policy name can't be empty"; got nil`) + } + p = Policy{ + Name: "dummy", + Type: "unknownDummyType", + } + if err := p.Valid(); err == nil { + t.Error(`want error "policy type unknown"; got nil`) + } +} + +func TestExecuteInvalidPolicy(t *testing.T) { + p := Policy{ + Name: "dummy", + Type: "unknownDummyType", + } + if _, err := p.Execute(context.TODO()); err == nil { + t.Error(`want error "policy type unknown"; got nil`) + } +} + +func TestExecute(t *testing.T) { + p := Policy{ + Name: "dummy", + Type: "fake", + M: map[string]string{ + "foo": "foo_value", + "interval": "200ms", + }, + } + ctx, cancel := context.WithCancel(context.Background()) + out, err := p.Execute(ctx) + if err != nil { + t.Error(err) + } + go func() { + time.Sleep(1 * time.Second) + cancel() + }() + + var count int + // Here, we are also able to test whether out is being + // closed when cancel() is called. If out is not closed, + // this test should have been running forever. + for evt := range out { + count++ + if evt.Data["foo"] != "foo_value" { + t.Errorf(`want evt.Data["foo"] = %s; got %s`, "foo", evt.Data["foo"]) + } + } + + // The interval for the dummy policy is 200ms. + // We are calling cancel after 1 sec. Typically, we receive + // 4 or 5 events in that duration. + if count != 4 && count != 5 { + t.Errorf(`want count to be either 4 or 5; got %d`, count) + } +}