From cad45edbc3d2c68744fb3762595a151266f04a0d Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 18 Aug 2015 16:17:49 +0530 Subject: [PATCH 01/59] publish metrics to a NATS channel instead of an HTTP endpoint --- cmd/recond/agent.go | 31 +++---------------------------- cmd/recond/main.go | 9 ++------- 2 files changed, 5 insertions(+), 35 deletions(-) diff --git a/cmd/recond/agent.go b/cmd/recond/agent.go index 0ac17a8..2486fea 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" @@ -62,33 +60,10 @@ func (a *Agent) register(addr string) error { return nil } -func (a *Agent) update(addr string) error { - var buf bytes.Buffer - - m := recon.Metric{ +func (a *Agent) update() { + m := &recon.Metric{ AgentUID: a.UID, Data: accumulateData(), } - - if err := json.NewEncoder(&buf).Encode(&m); err != nil { - return err - } - - l, err := url.Parse(addr + metricsAPIPath) - if err != nil { - return err - } - resp, err := http.Post(l.String(), "application/json", &buf) - 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 + natsEncConn.Publish("marksman_metrics", &m) } diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 3581bc4..ee050af 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -14,10 +14,7 @@ import ( "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. @@ -58,8 +55,6 @@ func main() { c := time.Tick(updateInterval) for now := range c { log.Println("Update sent at", now) - if err := agent.update(*marksmanAddr); err != nil { - log.Println(err) - } + agent.update() } } From 09b2c9a0381e55395ff22c5c49493645f6e6221f Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 19 Aug 2015 14:15:45 +0530 Subject: [PATCH 02/59] execute policies in the daemon --- cmd/recond/main.go | 7 +++++++ policy/policy.go | 33 ++++++++++++++++++++++++++++++ policy/tcp.go | 50 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+) create mode 100644 policy/policy.go create mode 100644 policy/tcp.go diff --git a/cmd/recond/main.go b/cmd/recond/main.go index ee050af..482401a 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -11,6 +11,7 @@ import ( "time" "github.com/codeignition/recon/cmd/recond/config" + "github.com/codeignition/recon/policy" "github.com/nats-io/nats" ) @@ -52,6 +53,12 @@ func main() { fmt.Printf("Received a message: %s\n", s) }) + natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { + fmt.Printf("Received a Policy: %v\n", p) + err := p.Execute() + natsEncConn.Publish(reply, err) + }) + c := time.Tick(updateInterval) for now := range c { log.Println("Update sent at", now) diff --git a/policy/policy.go b/policy/policy.go new file mode 100644 index 0000000..9bb5b3a --- /dev/null +++ b/policy/policy.go @@ -0,0 +1,33 @@ +// 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 + +// Type denotes the monitoring policy type. +// +// e.g. "tcp" +type Type string + +// Policy is the map containing the rules of a particular monitoring policy. +// +// e.g. "tcp" PolicyType requires 2 policy keys "port" and "frequency" +type Policy struct { + Name string `json:"name"` + AgentUID string `json:"agent_uid"` + Type Type `json:"policy_type"` + M map[string]string `json:"m"` +} + +// Config is the format used to encode/decode the monitoring policy +// received from the message queue or to store in the config file +type Config []Policy + +// PolicyFuncMap maps a PolicyType to a handler function +var PolicyFuncMap = map[Type]func(Policy) error{ + "tcp": tcpPolicyHandler, +} + +func (p Policy) Execute() error { + return PolicyFuncMap[p.Type](p) +} diff --git a/policy/tcp.go b/policy/tcp.go new file mode 100644 index 0000000..10c29b6 --- /dev/null +++ b/policy/tcp.go @@ -0,0 +1,50 @@ +package policy + +import ( + "errors" + "fmt" + "net" + "time" +) + +func tcpPolicyHandler(p Policy) error { + // Always use v, ok := p[key] form to avoid panic + port, ok := p.M["port"] + if !ok { + return errors.New(`"port" key missing in tcp policy`) + } + freq, ok := p.M["frequency"] + if !ok { + return errors.New(`"frequency" 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". + d, err := time.ParseDuration(freq) + if err != nil { + return err + } + + // This check is here to ensure time.Ticker(d) doesn't panic + if d <= 0 { + return errors.New("frequency must be a positive quantity") + } + go func() { + for now := range time.Tick(d) { + _, err := net.DialTimeout("tcp", port, d) + if err != nil { + fmt.Println(now, p.Name, "failure") + // TODO: sendErrorToMarksman(Agent, Policy, err) + } else { + fmt.Println(now, p.Name, "success") + // TODO: sendSuccessToMarksman(Agent, Policy, err) + } + + } + }() + return nil +} From 509410760d5c330bd45196d716602f2edc8c59c5 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 19 Aug 2015 14:17:10 +0530 Subject: [PATCH 03/59] add comment on valid time units --- policy/tcp.go | 1 + 1 file changed, 1 insertion(+) diff --git a/policy/tcp.go b/policy/tcp.go index 10c29b6..b7e9ba5 100644 --- a/policy/tcp.go +++ b/policy/tcp.go @@ -24,6 +24,7 @@ func tcpPolicyHandler(p Policy) error { // 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(freq) if err != nil { return err From b74ef149645ff9563f3932418ebe463166091400 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 11:15:51 +0530 Subject: [PATCH 04/59] update readme --- README.md | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index ef80038..ba73161 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). + +[`marksman`](https://github.com/codeignition/marksman) is the master server that aggregates the metrics from agents and exposes a public HTTP API. + +#### License BSD 3-clause "New" or "Revised" license -### Disclaimer +#### Disclaimer So far the project is tested only on Linux, specifically Ubuntu 14.04. -### About - -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. +#### Contributing +Check out the code and jump to Issues section to join a conversation or to start one for feedback or questions on the code, features and anything else. From 3ce9cc06e05fe50183f0c22d7b6197df13d1a3fb Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 12:03:03 +0530 Subject: [PATCH 05/59] acknowledge policy receive using a string message error type was not recognized by nats Encoder --- cmd/recond/main.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 482401a..9fa47f2 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -56,7 +56,11 @@ func main() { natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { fmt.Printf("Received a Policy: %v\n", p) err := p.Execute() - natsEncConn.Publish(reply, err) + if err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + natsEncConn.Publish(reply, "policy ack") // acknowledge }) c := time.Tick(updateInterval) From 09811f687d1bd3d067ae8148d1d8d52cad512657 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 12:04:55 +0530 Subject: [PATCH 06/59] add policy validation --- policy/policy.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/policy/policy.go b/policy/policy.go index 9bb5b3a..b1467ea 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -4,6 +4,8 @@ package policy +import "errors" + // Type denotes the monitoring policy type. // // e.g. "tcp" @@ -29,5 +31,21 @@ var PolicyFuncMap = map[Type]func(Policy) error{ } func (p Policy) Execute() error { + if err := p.Valid(); err != nil { + return err + } return PolicyFuncMap[p.Type](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") + } + // TODO: also check if the name conflicts with already + // existing ones + if _, ok := PolicyFuncMap[p.Type]; !ok { + return errors.New("policy type unknown") + } + return nil +} From d731a984c622f1b8ab1f8a6419bb5fb399ac0326 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 12:50:05 +0530 Subject: [PATCH 07/59] save policy config on policy receive --- cmd/recond/config/config.go | 10 +++++++++- cmd/recond/main.go | 8 ++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 6c9fcb1..8375ca8 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -17,6 +17,7 @@ import ( "path/filepath" "github.com/codeignition/recon/internal/fileutil" + "github.com/codeignition/recon/policy" ) const configFileName = ".recond.json" @@ -35,7 +36,8 @@ 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 + UID string `json:"uid"` // Unique Identifier to register with marksman + PolicyConfig policy.Config } // Init initializes and returns a Config. i.e. if the config file doesn't exist, @@ -90,6 +92,12 @@ func (c *Config) Save() error { return nil } +func (c *Config) AddPolicy(p policy.Policy) { + c.PolicyConfig = append(c.PolicyConfig, p) + // TODO: possible race condition. Use a mutex lock while + // writing to the slice PolicyConfig. +} + // 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/main.go b/cmd/recond/main.go index 9fa47f2..52a38e1 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -55,8 +55,12 @@ func main() { natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { fmt.Printf("Received a Policy: %v\n", p) - err := p.Execute() - if err != nil { + if err := p.Execute(); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + conf.AddPolicy(*p) + if err := conf.Save(); err != nil { natsEncConn.Publish(reply, err.Error()) return } From d9849f6458c0601cf0dd7037636f061f98de145c Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 12:55:43 +0530 Subject: [PATCH 08/59] remove json tags in config they were unnecessary --- cmd/recond/config/config.go | 2 +- cmd/recond/config/config_test.go | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 8375ca8..b564f51 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -36,7 +36,7 @@ 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 + UID string // Unique Identifier to register with marksman PolicyConfig policy.Config } diff --git a/cmd/recond/config/config_test.go b/cmd/recond/config/config_test.go index 215ad9b..6509cf9 100644 --- a/cmd/recond/config/config_test.go +++ b/cmd/recond/config/config_test.go @@ -31,7 +31,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 +49,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 +94,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 { From 09fc5df9807a75ea00f94ff7246dc65463127c77 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 13:19:23 +0530 Subject: [PATCH 09/59] remove json tags for Policy --- policy/policy.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index b1467ea..2b52862 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -15,10 +15,10 @@ type Type string // // e.g. "tcp" PolicyType requires 2 policy keys "port" and "frequency" type Policy struct { - Name string `json:"name"` - AgentUID string `json:"agent_uid"` - Type Type `json:"policy_type"` - M map[string]string `json:"m"` + Name string + AgentUID string + Type Type + M map[string]string } // Config is the format used to encode/decode the monitoring policy From c6b19d06f812c6db6631ea4637188831c3c52f60 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 13:19:50 +0530 Subject: [PATCH 10/59] indent json in config file for readability --- cmd/recond/config/config.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index b564f51..711cc23 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -7,6 +7,7 @@ package config import ( + "bytes" "crypto/rand" "encoding/json" "fmt" @@ -80,10 +81,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 From 16589560cde1e9e831ca181d6d006df19422223d Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 14:19:50 +0530 Subject: [PATCH 11/59] check whether a policy with the same name exists while adding Also modify the config before the policy starts executing --- cmd/recond/config/config.go | 9 ++++++++- cmd/recond/main.go | 7 +++++-- policy/policy.go | 2 -- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 711cc23..c3be205 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -10,6 +10,7 @@ import ( "bytes" "crypto/rand" "encoding/json" + "errors" "fmt" "io" "log" @@ -96,10 +97,16 @@ func (c *Config) Save() error { return nil } -func (c *Config) AddPolicy(p policy.Policy) { +func (c *Config) AddPolicy(p policy.Policy) error { + 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) // TODO: possible race condition. Use a mutex lock while // writing to the slice PolicyConfig. + return nil } // parseConfig reads from a io.Reader and diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 52a38e1..08ede0c 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -55,15 +55,18 @@ func main() { natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { fmt.Printf("Received a Policy: %v\n", p) - if err := p.Execute(); err != nil { + if err := conf.AddPolicy(*p); err != nil { natsEncConn.Publish(reply, err.Error()) return } - conf.AddPolicy(*p) if err := conf.Save(); err != nil { natsEncConn.Publish(reply, err.Error()) return } + if err := p.Execute(); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } natsEncConn.Publish(reply, "policy ack") // acknowledge }) diff --git a/policy/policy.go b/policy/policy.go index 2b52862..233a887 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -42,8 +42,6 @@ func (p Policy) Valid() error { if p.Name == "" { return errors.New("policy name can't be empty") } - // TODO: also check if the name conflicts with already - // existing ones if _, ok := PolicyFuncMap[p.Type]; !ok { return errors.New("policy type unknown") } From d7ba3d4f52ed8371e395208771114653ace76eae Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 14:49:17 +0530 Subject: [PATCH 12/59] add contributing document --- CONTRIBUTING.md | 26 ++++++++++++++++++++++++++ README.md | 2 +- 2 files changed, 27 insertions(+), 1 deletion(-) create mode 100644 CONTRIBUTING.md diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..310a1e5 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,26 @@ +# 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/) + +### Documentation + +Do not wrap lines at 80 characters. + +### 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 ba73161..fae5e34 100644 --- a/README.md +++ b/README.md @@ -35,4 +35,4 @@ So far the project is tested only on Linux, specifically Ubuntu 14.04. #### Contributing -Check out the code and jump to Issues section to join a conversation or to start one for feedback or questions on the code, features and anything else. +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. From 7f9a147a53de13406cb39fcca658b3e2fb678880 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 14:51:03 +0530 Subject: [PATCH 13/59] remove unnecessary documentation section in contributing.md --- CONTRIBUTING.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 310a1e5..342baf1 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -17,10 +17,6 @@ Please file bugs in the issues section detailing the problem completely. 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/) -### Documentation - -Do not wrap lines at 80 characters. - ### License By contributing to Recon, you agree that your contributions will be licensed under its BSD license. \ No newline at end of file From 75932c42d3033af0e3506d5c6f9162a79bfa80e8 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 25 Aug 2015 15:14:27 +0530 Subject: [PATCH 14/59] fix nit in readme --- README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index fae5e34..bfedd4b 100644 --- a/README.md +++ b/README.md @@ -25,10 +25,6 @@ go get github.com/codeignition/recon/... [`marksman`](https://github.com/codeignition/marksman) is the master server that aggregates the metrics from agents and exposes a public HTTP API. -#### License - -BSD 3-clause "New" or "Revised" license - #### Disclaimer So far the project is tested only on Linux, specifically Ubuntu 14.04. @@ -36,3 +32,7 @@ So far the project is tested only on Linux, specifically Ubuntu 14.04. #### 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. + +#### License + +BSD 3-clause "New" or "Revised" license \ No newline at end of file From 127bc4c6336146a880a11e2b2d8c958a54dc491e Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 26 Aug 2015 16:53:37 +0530 Subject: [PATCH 15/59] run stored policies on start, fix #2 --- cmd/recond/main.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 08ede0c..e108cfc 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -49,6 +49,8 @@ func main() { defer natsEncConn.Close() + go runStoredPolicies(conf) + natsEncConn.Subscribe(agent.UID, func(s string) { fmt.Printf("Received a message: %s\n", s) }) @@ -76,3 +78,13 @@ func main() { agent.update() } } + +func runStoredPolicies(c *config.Config) { + for _, p := range c.PolicyConfig { + go func() { + if err := p.Execute(); err != nil { + log.Fatal(err) + } + }() + } +} From 0d34b9ac78e497204afc147098b157ce71087224 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 27 Aug 2015 14:24:12 +0530 Subject: [PATCH 16/59] Use asynchronous API for policy handlers, fix #3 --- cmd/recond/main.go | 14 +++++++++++--- policy/event.go | 14 ++++++++++++++ policy/policy.go | 6 +++--- policy/tcp.go | 36 ++++++++++++++++++++++++------------ 4 files changed, 52 insertions(+), 18 deletions(-) create mode 100644 policy/event.go diff --git a/cmd/recond/main.go b/cmd/recond/main.go index e108cfc..aaf5144 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -65,11 +65,15 @@ func main() { natsEncConn.Publish(reply, err.Error()) return } - if err := p.Execute(); err != nil { + events, err := p.Execute() + if err != nil { natsEncConn.Publish(reply, err.Error()) return } natsEncConn.Publish(reply, "policy ack") // acknowledge + for e := range events { + fmt.Printf("%+v\n", e) + } }) c := time.Tick(updateInterval) @@ -82,8 +86,12 @@ func main() { func runStoredPolicies(c *config.Config) { for _, p := range c.PolicyConfig { go func() { - if err := p.Execute(); err != nil { - log.Fatal(err) + events, err := p.Execute() + if err != nil { + log.Fatal(err) // TODO: send to a nats errors channel + } + for e := range events { + fmt.Printf("%+v\n", e) } }() } diff --git a/policy/event.go b/policy/event.go new file mode 100644 index 0000000..3d94671 --- /dev/null +++ b/policy/event.go @@ -0,0 +1,14 @@ +// 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 + Policy Policy + Data map[string]interface{} // Data may include status, stats, etc. +} diff --git a/policy/policy.go b/policy/policy.go index 233a887..81fcfe4 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -26,13 +26,13 @@ type Policy struct { type Config []Policy // PolicyFuncMap maps a PolicyType to a handler function -var PolicyFuncMap = map[Type]func(Policy) error{ +var PolicyFuncMap = map[Type]func(Policy) (<-chan Event, error){ "tcp": tcpPolicyHandler, } -func (p Policy) Execute() error { +func (p Policy) Execute() (<-chan Event, error) { if err := p.Valid(); err != nil { - return err + return nil, err } return PolicyFuncMap[p.Type](p) } diff --git a/policy/tcp.go b/policy/tcp.go index b7e9ba5..38ce238 100644 --- a/policy/tcp.go +++ b/policy/tcp.go @@ -2,20 +2,19 @@ package policy import ( "errors" - "fmt" "net" "time" ) -func tcpPolicyHandler(p Policy) error { +func tcpPolicyHandler(p Policy) (<-chan Event, error) { // Always use v, ok := p[key] form to avoid panic port, ok := p.M["port"] if !ok { - return errors.New(`"port" key missing in tcp policy`) + return nil, errors.New(`"port" key missing in tcp policy`) } freq, ok := p.M["frequency"] if !ok { - return errors.New(`"frequency" key missing in tcp policy`) + return nil, errors.New(`"frequency" key missing in tcp policy`) } // From the time package docs: @@ -27,25 +26,38 @@ func tcpPolicyHandler(p Policy) error { // Valid time units are "ns", "us", "ms", "s", "m", "h". d, err := time.ParseDuration(freq) if err != nil { - return err + return nil, err } // This check is here to ensure time.Ticker(d) doesn't panic if d <= 0 { - return errors.New("frequency must be a positive quantity") + return nil, errors.New("frequency must be a positive quantity") } + + out := make(chan Event) go func() { for now := range time.Tick(d) { _, err := net.DialTimeout("tcp", port, d) if err != nil { - fmt.Println(now, p.Name, "failure") - // TODO: sendErrorToMarksman(Agent, Policy, err) + out <- Event{ + Time: now, + Policy: p, + Data: map[string]interface{}{ + "status": "failure", + "error": err, + }, + } } else { - fmt.Println(now, p.Name, "success") - // TODO: sendSuccessToMarksman(Agent, Policy, err) + out <- Event{ + Time: now, + Policy: p, + Data: map[string]interface{}{ + "status": "success", + }, + } } - + // TODO: think about when to close the out channel . } }() - return nil + return out, nil } From 1e79680289b67f271900d647d71367d6cde71d58 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 27 Aug 2015 15:41:43 +0530 Subject: [PATCH 17/59] unexport policyFuncMap so that package consumers can't change it --- policy/policy.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index 81fcfe4..726ab50 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -25,8 +25,8 @@ type Policy struct { // received from the message queue or to store in the config file type Config []Policy -// PolicyFuncMap maps a PolicyType to a handler function -var PolicyFuncMap = map[Type]func(Policy) (<-chan Event, error){ +// policyFuncMap maps a PolicyType to a handler function +var policyFuncMap = map[Type]func(Policy) (<-chan Event, error){ "tcp": tcpPolicyHandler, } @@ -34,7 +34,7 @@ func (p Policy) Execute() (<-chan Event, error) { if err := p.Valid(); err != nil { return nil, err } - return PolicyFuncMap[p.Type](p) + return policyFuncMap[p.Type](p) } // Valid checks whether the policy is valid. @@ -42,7 +42,7 @@ func (p Policy) Valid() error { if p.Name == "" { return errors.New("policy name can't be empty") } - if _, ok := PolicyFuncMap[p.Type]; !ok { + if _, ok := policyFuncMap[p.Type]; !ok { return errors.New("policy type unknown") } return nil From e59b56ca82ee726b74caa3dfc984f22fe130545f Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 27 Aug 2015 15:44:48 +0530 Subject: [PATCH 18/59] add context.Context in the function signature for policy handlers to allow cancelation and may be timeouts --- policy/policy.go | 10 +++++++--- policy/tcp.go | 4 +++- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index 726ab50..e1d9278 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -4,7 +4,11 @@ package policy -import "errors" +import ( + "errors" + + "golang.org/x/net/context" +) // Type denotes the monitoring policy type. // @@ -26,7 +30,7 @@ type Policy struct { type Config []Policy // policyFuncMap maps a PolicyType to a handler function -var policyFuncMap = map[Type]func(Policy) (<-chan Event, error){ +var policyFuncMap = map[Type]func(context.Context, Policy) (<-chan Event, error){ "tcp": tcpPolicyHandler, } @@ -34,7 +38,7 @@ func (p Policy) Execute() (<-chan Event, error) { if err := p.Valid(); err != nil { return nil, err } - return policyFuncMap[p.Type](p) + return policyFuncMap[p.Type](context.TODO(), p) } // Valid checks whether the policy is valid. diff --git a/policy/tcp.go b/policy/tcp.go index 38ce238..d5ab570 100644 --- a/policy/tcp.go +++ b/policy/tcp.go @@ -4,9 +4,11 @@ import ( "errors" "net" "time" + + "golang.org/x/net/context" ) -func tcpPolicyHandler(p Policy) (<-chan Event, error) { +func tcpPolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { // Always use v, ok := p[key] form to avoid panic port, ok := p.M["port"] if !ok { From f70be5f713e299cd054dbc5e2d09295eb4b34068 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 11:41:08 +0530 Subject: [PATCH 19/59] Add HandlerFunc type with handler function signature Also add NewHandler function --- policy/policy.go | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/policy/policy.go b/policy/policy.go index e1d9278..599fc1f 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -29,8 +29,10 @@ type Policy struct { // received from the message queue or to store in the config file type Config []Policy +type HandlerFunc func(context.Context, Policy) (<-chan Event, error) + // policyFuncMap maps a PolicyType to a handler function -var policyFuncMap = map[Type]func(context.Context, Policy) (<-chan Event, error){ +var policyFuncMap = map[Type]HandlerFunc{ "tcp": tcpPolicyHandler, } @@ -51,3 +53,15 @@ func (p Policy) Valid() error { } return nil } + +func NewHandler(policyType Type, handlerFunc HandlerFunc) error { + if policyType == "" { + return errors.New("policy type can't be empty") + } + + if _, ok := policyFuncMap[policyType]; ok { + return errors.New("handler for the policy type already exists") + } + policyFuncMap[policyType] = handlerFunc + return nil +} From 9a143fd0d8a53b2a84012e93e236d48ea99310b4 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 11:42:16 +0530 Subject: [PATCH 20/59] add tests for policy package --- policy/policy_test.go | 74 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 policy/policy_test.go diff --git a/policy/policy_test.go b/policy/policy_test.go new file mode 100644 index 0000000..5a86cc5 --- /dev/null +++ b/policy/policy_test.go @@ -0,0 +1,74 @@ +// 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`) + } + bar, ok := p.M["bar"] + if !ok { + return nil, errors.New(`"bar" key missing in fake policy`) + } + out := make(chan Event) + go func() { + out <- Event{ + Time: time.Now(), + Policy: p, + Data: map[string]interface{}{ + "count": 1, + "foo": foo, + "bar": bar, + }, + } + out <- Event{ + Time: time.Now(), + Policy: p, + Data: map[string]interface{}{ + "count": 2, + "foo": foo, + "bar": bar, + }, + } + close(out) + }() + return out, nil +} + +func TestNewHandler(t *testing.T) { + err := NewHandler("", fakePolicyHandler) + if err == nil { + t.Fatal(errors.New("NewHandler should return an error when the type is empty")) + } + err = NewHandler("fake", fakePolicyHandler) + if err != nil { + t.Fatal(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`) + } +} From e1f4380c67d0f834e62fa78ff9ea90a707981e88 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 12:08:45 +0530 Subject: [PATCH 21/59] policy: add tests for execute --- policy/policy_test.go | 38 ++++++++++++++++++++++++++++++++------ 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/policy/policy_test.go b/policy/policy_test.go index 5a86cc5..35d1ec9 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -27,18 +27,16 @@ func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { Time: time.Now(), Policy: p, Data: map[string]interface{}{ - "count": 1, - "foo": foo, - "bar": bar, + "foo": foo, + "bar": bar, }, } out <- Event{ Time: time.Now(), Policy: p, Data: map[string]interface{}{ - "count": 2, - "foo": foo, - "bar": bar, + "foo": foo, + "bar": bar, }, } close(out) @@ -72,3 +70,31 @@ func TestValid(t *testing.T) { 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", + "bar": "bar_value", + }, + } + out, err := p.Execute() + if err != nil { + t.Error(err) + } + var count int + for evt := range out { + count++ + if evt.Data["foo"] != "foo_value" { + t.Errorf(`want evt.Data["foo"] = %s; got %s`, "foo", evt.Data["foo"]) + } + if evt.Data["bar"] != "bar_value" { + t.Errorf(`want evt.Data["bar"] = %s; got %s`, "bar", evt.Data["bar"]) + } + } + if count != 2 { + t.Errorf(`expected %d events; got %d events`, 2, count) + } +} From 476043ec0aaefadc61f751bcdf38d229f9832715 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 12:13:49 +0530 Subject: [PATCH 22/59] policy: pass context.Context into Execute --- cmd/recond/main.go | 6 ++++-- policy/policy.go | 4 ++-- policy/policy_test.go | 2 +- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index aaf5144..cff9780 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -10,6 +10,8 @@ import ( "log" "time" + "golang.org/x/net/context" + "github.com/codeignition/recon/cmd/recond/config" "github.com/codeignition/recon/policy" "github.com/nats-io/nats" @@ -65,7 +67,7 @@ func main() { natsEncConn.Publish(reply, err.Error()) return } - events, err := p.Execute() + events, err := p.Execute(context.TODO()) if err != nil { natsEncConn.Publish(reply, err.Error()) return @@ -86,7 +88,7 @@ func main() { func runStoredPolicies(c *config.Config) { for _, p := range c.PolicyConfig { go func() { - events, err := p.Execute() + events, err := p.Execute(context.TODO()) if err != nil { log.Fatal(err) // TODO: send to a nats errors channel } diff --git a/policy/policy.go b/policy/policy.go index 599fc1f..3fd5335 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -36,11 +36,11 @@ var policyFuncMap = map[Type]HandlerFunc{ "tcp": tcpPolicyHandler, } -func (p Policy) Execute() (<-chan Event, error) { +func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { if err := p.Valid(); err != nil { return nil, err } - return policyFuncMap[p.Type](context.TODO(), p) + return policyFuncMap[p.Type](ctx, p) } // Valid checks whether the policy is valid. diff --git a/policy/policy_test.go b/policy/policy_test.go index 35d1ec9..aeda5df 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -80,7 +80,7 @@ func TestExecute(t *testing.T) { "bar": "bar_value", }, } - out, err := p.Execute() + out, err := p.Execute(context.TODO()) if err != nil { t.Error(err) } From 7b46448fb99f9c98895e49f9877f09d774807f3b Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 13:08:20 +0530 Subject: [PATCH 23/59] policy: test stopping policy execution using context cancel --- policy/policy_test.go | 72 +++++++++++++++++++++++++++---------------- 1 file changed, 46 insertions(+), 26 deletions(-) diff --git a/policy/policy_test.go b/policy/policy_test.go index aeda5df..859a120 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -17,29 +17,39 @@ func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { if !ok { return nil, errors.New(`"foo" key missing in fake policy`) } - bar, ok := p.M["bar"] + interval, ok := p.M["interval"] if !ok { - return nil, errors.New(`"bar" key missing in fake policy`) + 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("frequency must be a positive quantity") + } + out := make(chan Event) go func() { - out <- Event{ - Time: time.Now(), - Policy: p, - Data: map[string]interface{}{ - "foo": foo, - "bar": bar, - }, - } - out <- Event{ - Time: time.Now(), - Policy: p, - Data: map[string]interface{}{ - "foo": foo, - "bar": bar, - }, + 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, + }, + } + } } - close(out) }() return out, nil } @@ -76,25 +86,35 @@ func TestExecute(t *testing.T) { Name: "dummy", Type: "fake", M: map[string]string{ - "foo": "foo_value", - "bar": "bar_value", + "foo": "foo_value", + "interval": "200ms", }, } - out, err := p.Execute(context.TODO()) + 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"]) } - if evt.Data["bar"] != "bar_value" { - t.Errorf(`want evt.Data["bar"] = %s; got %s`, "bar", evt.Data["bar"]) - } } - if count != 2 { - t.Errorf(`expected %d events; got %d events`, 2, count) + + // 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) } } From 6580a5929ac0ae3b934a4048e4ecac24d3c6bffa Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 13:24:28 +0530 Subject: [PATCH 24/59] fix oversight in the test error message --- policy/policy_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/policy/policy_test.go b/policy/policy_test.go index 859a120..1d0c4e0 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -28,7 +28,7 @@ func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { // This check is here to ensure time.Ticker(d) doesn't panic if d <= 0 { - return nil, errors.New("frequency must be a positive quantity") + return nil, errors.New("interval must be a positive quantity") } out := make(chan Event) From d9d76d733279d02a0ea0de73f37b43027f4e2b10 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 14:48:26 +0530 Subject: [PATCH 25/59] policy: move tcp handler to a separate package the handlers subpackage in policy contains common handlers that we want to support. The consumer (recond) will have to underscore import handlers for its sideeffect of registering handlers --- policy/handlers/handlers.go | 11 +++++++++++ policy/{ => handlers}/tcp.go | 16 +++++++++++----- policy/policy.go | 4 +--- 3 files changed, 23 insertions(+), 8 deletions(-) create mode 100644 policy/handlers/handlers.go rename policy/{ => handlers}/tcp.go (77%) diff --git a/policy/handlers/handlers.go b/policy/handlers/handlers.go new file mode 100644 index 0000000..909f354 --- /dev/null +++ b/policy/handlers/handlers.go @@ -0,0 +1,11 @@ +// 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.NewHandler("tcp", TCP) +} diff --git a/policy/tcp.go b/policy/handlers/tcp.go similarity index 77% rename from policy/tcp.go rename to policy/handlers/tcp.go index d5ab570..38bb0b5 100644 --- a/policy/tcp.go +++ b/policy/handlers/tcp.go @@ -1,14 +1,20 @@ -package policy +// 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 tcpPolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { +func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { // Always use v, ok := p[key] form to avoid panic port, ok := p.M["port"] if !ok { @@ -36,12 +42,12 @@ func tcpPolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { return nil, errors.New("frequency must be a positive quantity") } - out := make(chan Event) + out := make(chan policy.Event) go func() { for now := range time.Tick(d) { _, err := net.DialTimeout("tcp", port, d) if err != nil { - out <- Event{ + out <- policy.Event{ Time: now, Policy: p, Data: map[string]interface{}{ @@ -50,7 +56,7 @@ func tcpPolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { }, } } else { - out <- Event{ + out <- policy.Event{ Time: now, Policy: p, Data: map[string]interface{}{ diff --git a/policy/policy.go b/policy/policy.go index 3fd5335..a3e659b 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -32,9 +32,7 @@ type Config []Policy type HandlerFunc func(context.Context, Policy) (<-chan Event, error) // policyFuncMap maps a PolicyType to a handler function -var policyFuncMap = map[Type]HandlerFunc{ - "tcp": tcpPolicyHandler, -} +var policyFuncMap = map[Type]HandlerFunc{} func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { if err := p.Valid(); err != nil { From 135d8f728d219e4f338e2b83c7e1d8195a1bd05d Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 14:50:24 +0530 Subject: [PATCH 26/59] underscore import policy/handlers for sideeffects --- cmd/recond/main.go | 1 + 1 file changed, 1 insertion(+) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index cff9780..c15cf86 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -14,6 +14,7 @@ import ( "github.com/codeignition/recon/cmd/recond/config" "github.com/codeignition/recon/policy" + _ "github.com/codeignition/recon/policy/handlers" "github.com/nats-io/nats" ) From 39b65dae59a6f8b66f3f57d6329c2387abad40cd Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 14:54:15 +0530 Subject: [PATCH 27/59] policy: change NewHandler to RegisterHandler to be semantically correct --- policy/handlers/handlers.go | 2 +- policy/policy.go | 4 ++-- policy/policy_test.go | 6 +++--- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/policy/handlers/handlers.go b/policy/handlers/handlers.go index 909f354..74a1ef1 100644 --- a/policy/handlers/handlers.go +++ b/policy/handlers/handlers.go @@ -7,5 +7,5 @@ package handlers import "github.com/codeignition/recon/policy" func init() { - policy.NewHandler("tcp", TCP) + policy.RegisterHandler("tcp", TCP) } diff --git a/policy/policy.go b/policy/policy.go index a3e659b..d456528 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -32,7 +32,7 @@ type Config []Policy type HandlerFunc func(context.Context, Policy) (<-chan Event, error) // policyFuncMap maps a PolicyType to a handler function -var policyFuncMap = map[Type]HandlerFunc{} +var policyFuncMap = make(map[Type]HandlerFunc) func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { if err := p.Valid(); err != nil { @@ -52,7 +52,7 @@ func (p Policy) Valid() error { return nil } -func NewHandler(policyType Type, handlerFunc HandlerFunc) error { +func RegisterHandler(policyType Type, handlerFunc HandlerFunc) error { if policyType == "" { return errors.New("policy type can't be empty") } diff --git a/policy/policy_test.go b/policy/policy_test.go index 1d0c4e0..d1c7da1 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -54,12 +54,12 @@ func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { return out, nil } -func TestNewHandler(t *testing.T) { - err := NewHandler("", fakePolicyHandler) +func TestRegisterHandler(t *testing.T) { + err := RegisterHandler("", fakePolicyHandler) if err == nil { t.Fatal(errors.New("NewHandler should return an error when the type is empty")) } - err = NewHandler("fake", fakePolicyHandler) + err = RegisterHandler("fake", fakePolicyHandler) if err != nil { t.Fatal(err) } From dcf8f53d06ab607232c1151ee25f57171c1bf9b6 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 15:03:46 +0530 Subject: [PATCH 28/59] policy: remove unnecessary type "Type" --- policy/policy.go | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index d456528..ac47fd4 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -10,18 +10,13 @@ import ( "golang.org/x/net/context" ) -// Type denotes the monitoring policy type. -// -// e.g. "tcp" -type Type string - // Policy is the map containing the rules of a particular monitoring policy. // // e.g. "tcp" PolicyType requires 2 policy keys "port" and "frequency" type Policy struct { Name string AgentUID string - Type Type + Type string // Type denotes the monitoring policy type. e.g. "tcp" M map[string]string } @@ -31,8 +26,8 @@ type Config []Policy type HandlerFunc func(context.Context, Policy) (<-chan Event, error) -// policyFuncMap maps a PolicyType to a handler function -var policyFuncMap = make(map[Type]HandlerFunc) +// policyFuncMap maps a policy type to a handler function +var policyFuncMap = make(map[string]HandlerFunc) func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { if err := p.Valid(); err != nil { @@ -52,7 +47,7 @@ func (p Policy) Valid() error { return nil } -func RegisterHandler(policyType Type, handlerFunc HandlerFunc) error { +func RegisterHandler(policyType string, handlerFunc HandlerFunc) error { if policyType == "" { return errors.New("policy type can't be empty") } From aea76e23c8420c59660e9a11b2ab23e5d01b4200 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 15:16:50 +0530 Subject: [PATCH 29/59] policy: add more tests --- policy/policy_test.go | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/policy/policy_test.go b/policy/policy_test.go index d1c7da1..15548da 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -63,6 +63,12 @@ func TestRegisterHandler(t *testing.T) { 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 TestValid(t *testing.T) { @@ -70,14 +76,24 @@ func TestValid(t *testing.T) { Name: "", } if err := p.Valid(); err == nil { - t.Error(`want error "policy name can't be empty" got 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`) + 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`) } } From cbb396e7a761c5ce4d5165fb3245c8d72f6dc70f Mon Sep 17 00:00:00 2001 From: Hari haran Date: Fri, 28 Aug 2015 20:16:38 +0530 Subject: [PATCH 30/59] policy: fix t.Fatal call --- policy/policy_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/policy/policy_test.go b/policy/policy_test.go index 15548da..e4da801 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -57,7 +57,7 @@ func fakePolicyHandler(ctx context.Context, p Policy) (<-chan Event, error) { func TestRegisterHandler(t *testing.T) { err := RegisterHandler("", fakePolicyHandler) if err == nil { - t.Fatal(errors.New("NewHandler should return an error when the type is empty")) + t.Fatal(`NewHandler should return an error when the type is empty`) } err = RegisterHandler("fake", fakePolicyHandler) if err != nil { From 0ef6f6f55b33d95fd6808bbdb1ed3b6995f56a94 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 12:24:08 +0530 Subject: [PATCH 31/59] recond: publish events instead of printing --- cmd/recond/main.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index c15cf86..f88f9f7 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -75,7 +75,7 @@ func main() { } natsEncConn.Publish(reply, "policy ack") // acknowledge for e := range events { - fmt.Printf("%+v\n", e) + natsEncConn.Publish("policy_events", e) } }) @@ -94,7 +94,7 @@ func runStoredPolicies(c *config.Config) { log.Fatal(err) // TODO: send to a nats errors channel } for e := range events { - fmt.Printf("%+v\n", e) + natsEncConn.Publish("policy_events", e) } }() } From 43c9f4df8d2446c18ad5b7d4aaf4959d33eddd1d Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 14:19:11 +0530 Subject: [PATCH 32/59] add system_data policy handler --- policy/handlers/handlers.go | 1 + policy/handlers/systemdata.go | 74 +++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+) create mode 100644 policy/handlers/systemdata.go diff --git a/policy/handlers/handlers.go b/policy/handlers/handlers.go index 74a1ef1..1a7f757 100644 --- a/policy/handlers/handlers.go +++ b/policy/handlers/handlers.go @@ -8,4 +8,5 @@ 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..09745f3 --- /dev/null +++ b/policy/handlers/systemdata.go @@ -0,0 +1,74 @@ +// 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" + "os/user" + "time" + + "github.com/codeignition/recon/metrics/netstat" + "github.com/codeignition/recon/metrics/ps" + "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(), + Policy: p, + Data: accumulateSystemData(), + } + } + } + }() + return out, nil +} + +func accumulateSystemData() 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 +} From 31d2d59c53587dd2641f7bd0bca22c973d297641 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 14:22:24 +0530 Subject: [PATCH 33/59] rename types.go to agent.go as recon.Metric type is removed --- types.go => agent.go | 8 -------- 1 file changed, 8 deletions(-) rename types.go => agent.go (65%) diff --git a/types.go b/agent.go similarity index 65% rename from types.go rename to agent.go index 6c6f503..ba1797b 100644 --- a/types.go +++ b/agent.go @@ -9,11 +9,3 @@ package recon 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"` -} From a442c4cb60c38675e64160b81476993060617e69 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 14:24:54 +0530 Subject: [PATCH 34/59] remove system data handling as it is abstracted into a default policy --- cmd/recond/accumulate.go | 43 ---------------------------------------- cmd/recond/agent.go | 8 -------- cmd/recond/main.go | 43 ++++++++++++++++++++++++++++++---------- 3 files changed, 33 insertions(+), 61 deletions(-) delete mode 100644 cmd/recond/accumulate.go 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 2486fea..ebe9c7c 100644 --- a/cmd/recond/agent.go +++ b/cmd/recond/agent.go @@ -59,11 +59,3 @@ func (a *Agent) register(addr string) error { } return nil } - -func (a *Agent) update() { - m := &recon.Metric{ - AgentUID: a.UID, - Data: accumulateData(), - } - natsEncConn.Publish("marksman_metrics", &m) -} diff --git a/cmd/recond/main.go b/cmd/recond/main.go index f88f9f7..effd2ea 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -8,7 +8,6 @@ import ( "flag" "fmt" "log" - "time" "golang.org/x/net/context" @@ -24,10 +23,6 @@ const agentsAPIPath = "/api/agents" // agents path in the marksman server // 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 - func main() { log.SetPrefix("recond: ") @@ -52,6 +47,10 @@ func main() { defer natsEncConn.Close() + if err := addSystemDataPolicy(conf); err != nil { + log.Fatal(err) + } + go runStoredPolicies(conf) natsEncConn.Subscribe(agent.UID, func(s string) { @@ -79,11 +78,9 @@ func main() { } }) - c := time.Tick(updateInterval) - for now := range c { - log.Println("Update sent at", now) - agent.update() - } + // this is just to block the main function from exiting + c := make(chan struct{}) + <-c } func runStoredPolicies(c *config.Config) { @@ -99,3 +96,29 @@ func runStoredPolicies(c *config.Config) { }() } } + +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 +} From 66bd4a94a8961b19c5450e27214aec1781c1ec61 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 14:53:25 +0530 Subject: [PATCH 35/59] change tcp policy keys and use ctx to close events channel --- policy/handlers/tcp.go | 57 ++++++++++++++++++++++++------------------ 1 file changed, 32 insertions(+), 25 deletions(-) diff --git a/policy/handlers/tcp.go b/policy/handlers/tcp.go index 38bb0b5..cb151d3 100644 --- a/policy/handlers/tcp.go +++ b/policy/handlers/tcp.go @@ -16,13 +16,13 @@ import ( func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { // Always use v, ok := p[key] form to avoid panic - port, ok := p.M["port"] + addr, ok := p.M["address"] if !ok { - return nil, errors.New(`"port" key missing in tcp policy`) + return nil, errors.New(`"address" key missing in tcp policy`) } - freq, ok := p.M["frequency"] + interval, ok := p.M["interval"] if !ok { - return nil, errors.New(`"frequency" key missing in tcp policy`) + return nil, errors.New(`"interval" key missing in tcp policy`) } // From the time package docs: @@ -32,39 +32,46 @@ func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { // 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(freq) + 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("frequency must be a positive quantity") + return nil, errors.New("interval must be a positive quantity") } out := make(chan policy.Event) go func() { - for now := range time.Tick(d) { - _, err := net.DialTimeout("tcp", port, d) - if err != nil { - out <- policy.Event{ - Time: now, - Policy: p, - Data: map[string]interface{}{ - "status": "failure", - "error": err, - }, - } - } else { - out <- policy.Event{ - Time: now, - Policy: p, - Data: map[string]interface{}{ - "status": "success", - }, + 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(), + Policy: p, + Data: map[string]interface{}{ + "status": "failure", + "error": err.Error(), + }, + } + } else { + out <- policy.Event{ + Time: time.Now(), + Policy: p, + Data: map[string]interface{}{ + "status": "success", + }, + } } } - // TODO: think about when to close the out channel . } }() return out, nil From 56f7d5c0f888bbd1c8e28609561bfa14fedc7d51 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 15:01:32 +0530 Subject: [PATCH 36/59] policy: fix comments and naming --- policy/policy.go | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index ac47fd4..4a4655a 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -10,30 +10,29 @@ import ( "golang.org/x/net/context" ) -// Policy is the map containing the rules of a particular monitoring policy. -// -// e.g. "tcp" PolicyType requires 2 policy keys "port" and "frequency" +// Policy represents a monitoring policy type Policy struct { - Name string - AgentUID string - Type string // Type denotes the monitoring policy type. e.g. "tcp" - M map[string]string + 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 the format used to encode/decode the monitoring policy -// received from the message queue or to store in the config file +// 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) -// policyFuncMap maps a policy type to a handler function -var policyFuncMap = make(map[string]HandlerFunc) +// handlerFuncMap maps a policy type to a handler function +var handlerFuncMap = make(map[string]HandlerFunc) func (p Policy) Execute(ctx context.Context) (<-chan Event, error) { if err := p.Valid(); err != nil { return nil, err } - return policyFuncMap[p.Type](ctx, p) + return handlerFuncMap[p.Type](ctx, p) } // Valid checks whether the policy is valid. @@ -41,7 +40,7 @@ func (p Policy) Valid() error { if p.Name == "" { return errors.New("policy name can't be empty") } - if _, ok := policyFuncMap[p.Type]; !ok { + if _, ok := handlerFuncMap[p.Type]; !ok { return errors.New("policy type unknown") } return nil @@ -52,9 +51,9 @@ func RegisterHandler(policyType string, handlerFunc HandlerFunc) error { return errors.New("policy type can't be empty") } - if _, ok := policyFuncMap[policyType]; ok { + if _, ok := handlerFuncMap[policyType]; ok { return errors.New("handler for the policy type already exists") } - policyFuncMap[policyType] = handlerFunc + handlerFuncMap[policyType] = handlerFunc return nil } From 94ee4c6295b6a6cb95eaf23599c9936818d89f90 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 15:17:58 +0530 Subject: [PATCH 37/59] policy: fix data race during concurrent RegisterHandler calls --- policy/policy.go | 30 +++++++++++++++++++++++++----- policy/policy_test.go | 16 ++++++++++++++++ 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/policy/policy.go b/policy/policy.go index 4a4655a..c9c7068 100644 --- a/policy/policy.go +++ b/policy/policy.go @@ -6,6 +6,7 @@ package policy import ( "errors" + "sync" "golang.org/x/net/context" ) @@ -26,13 +27,23 @@ type Config []Policy type HandlerFunc func(context.Context, Policy) (<-chan Event, error) // handlerFuncMap maps a policy type to a handler function -var handlerFuncMap = make(map[string]HandlerFunc) +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 } - return handlerFuncMap[p.Type](ctx, p) + + handlerFuncMap.Lock() + f := handlerFuncMap.m[p.Type] + handlerFuncMap.Unlock() + + return f(ctx, p) } // Valid checks whether the policy is valid. @@ -40,7 +51,12 @@ func (p Policy) Valid() error { if p.Name == "" { return errors.New("policy name can't be empty") } - if _, ok := handlerFuncMap[p.Type]; !ok { + + handlerFuncMap.Lock() + _, ok := handlerFuncMap.m[p.Type] + handlerFuncMap.Unlock() + + if !ok { return errors.New("policy type unknown") } return nil @@ -51,9 +67,13 @@ func RegisterHandler(policyType string, handlerFunc HandlerFunc) error { return errors.New("policy type can't be empty") } - if _, ok := handlerFuncMap[policyType]; ok { + handlerFuncMap.Lock() + defer handlerFuncMap.Unlock() + + if _, ok := handlerFuncMap.m[policyType]; ok { return errors.New("handler for the policy type already exists") } - handlerFuncMap[policyType] = handlerFunc + + handlerFuncMap.m[policyType] = handlerFunc return nil } diff --git a/policy/policy_test.go b/policy/policy_test.go index e4da801..7ac5411 100644 --- a/policy/policy_test.go +++ b/policy/policy_test.go @@ -71,6 +71,22 @@ func TestRegisterHandler(t *testing.T) { } } +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: "", From 05ac52c98f1d27f6354e1e0041fedf3513e290b8 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Mon, 31 Aug 2015 17:54:40 +0530 Subject: [PATCH 38/59] fix race condition while running stored policies --- cmd/recond/main.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index effd2ea..1dccc0c 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -84,16 +84,18 @@ func main() { } func runStoredPolicies(c *config.Config) { + log.Print("adding stored policies...") for _, p := range c.PolicyConfig { - go func() { + log.Print(p.Name) + go func(p policy.Policy) { events, err := p.Execute(context.TODO()) if err != nil { - log.Fatal(err) // TODO: send to a nats errors channel + log.Print(err) // TODO: send to a nats errors channel } for e := range events { natsEncConn.Publish("policy_events", e) } - }() + }(p) } } From ceaf0ead5ad76022dc3664d2f1ad059560b114c3 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 1 Sep 2015 15:32:14 +0530 Subject: [PATCH 39/59] log while adding policy --- cmd/recond/main.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 1dccc0c..d6ff915 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -84,9 +84,8 @@ func main() { } func runStoredPolicies(c *config.Config) { - log.Print("adding stored policies...") for _, p := range c.PolicyConfig { - log.Print(p.Name) + log.Printf("adding the policy %s...", p.Name) go func(p policy.Policy) { events, err := p.Execute(context.TODO()) if err != nil { From 08c1e202474e8738daff2bfb0bfb08f6d94130e3 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 1 Sep 2015 15:53:12 +0530 Subject: [PATCH 40/59] remove unnecessary channel subscription --- cmd/recond/main.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index d6ff915..f89619a 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -53,10 +53,6 @@ func main() { go runStoredPolicies(conf) - natsEncConn.Subscribe(agent.UID, func(s string) { - fmt.Printf("Received a message: %s\n", s) - }) - natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { fmt.Printf("Received a Policy: %v\n", p) if err := conf.AddPolicy(*p); err != nil { From cdea3e277519f94f6c95aa2cbf93fc5a60694502 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 1 Sep 2015 16:53:23 +0530 Subject: [PATCH 41/59] support cancelation of policies related to codeignition/marksman#3 --- cmd/recond/main.go | 37 +++++++++++++++++++++++++++++++++---- 1 file changed, 33 insertions(+), 4 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index f89619a..2f71a78 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -8,6 +8,7 @@ import ( "flag" "fmt" "log" + "sync" "golang.org/x/net/context" @@ -23,6 +24,15 @@ const agentsAPIPath = "/api/agents" // agents path in the marksman server // It is populated if the agent registers successfully. var natsEncConn *nats.EncodedConn +// 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), +} + func main() { log.SetPrefix("recond: ") @@ -53,7 +63,7 @@ func main() { go runStoredPolicies(conf) - natsEncConn.Subscribe(agent.UID+"_policy", func(subj, reply string, p *policy.Policy) { + natsEncConn.Subscribe(agent.UID+"_policy_add", func(subj, reply string, p *policy.Policy) { fmt.Printf("Received a Policy: %v\n", p) if err := conf.AddPolicy(*p); err != nil { natsEncConn.Publish(reply, err.Error()) @@ -63,17 +73,31 @@ func main() { natsEncConn.Publish(reply, err.Error()) return } - events, err := p.Execute(context.TODO()) + ctx, cancel := context.WithCancel(context.Background()) + events, err := p.Execute(ctx) if err != nil { natsEncConn.Publish(reply, err.Error()) return } - natsEncConn.Publish(reply, "policy ack") // acknowledge + ctxCancelFunc.Lock() + ctxCancelFunc.m[p.Name] = cancel + ctxCancelFunc.Unlock() + + natsEncConn.Publish(reply, "policy_add_ack") // acknowledge policy add for e := range events { natsEncConn.Publish("policy_events", e) } }) + natsEncConn.Subscribe(agent.UID+"_policy_delete", func(subj, reply string, p *policy.Policy) { + ctxCancelFunc.Lock() + cancel := ctxCancelFunc.m[p.Name] + // TODO: delete policy from conf and remove cancelfunc from map. + ctxCancelFunc.Unlock() + cancel() + natsEncConn.Publish(reply, "policy_delete_ack") // acknowledge policy delete + }) + // this is just to block the main function from exiting c := make(chan struct{}) <-c @@ -83,10 +107,15 @@ func runStoredPolicies(c *config.Config) { for _, p := range c.PolicyConfig { log.Printf("adding the policy %s...", p.Name) go func(p policy.Policy) { - events, err := p.Execute(context.TODO()) + 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) } From 37d11922809be2b9da7d55536dda43f2178e2ac4 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 1 Sep 2015 17:51:59 +0530 Subject: [PATCH 42/59] cleanup after policy delete --- cmd/recond/main.go | 32 +++++++++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 2f71a78..162ffa1 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -5,8 +5,8 @@ package main import ( + "errors" "flag" - "fmt" "log" "sync" @@ -64,7 +64,7 @@ func main() { go runStoredPolicies(conf) natsEncConn.Subscribe(agent.UID+"_policy_add", func(subj, reply string, p *policy.Policy) { - fmt.Printf("Received a Policy: %v\n", p) + log.Printf("policy_add received: %s\n", p.Name) if err := conf.AddPolicy(*p); err != nil { natsEncConn.Publish(reply, err.Error()) return @@ -90,11 +90,15 @@ func main() { }) natsEncConn.Subscribe(agent.UID+"_policy_delete", func(subj, reply string, p *policy.Policy) { + log.Printf("policy_delete received: %s\n", p.Name) ctxCancelFunc.Lock() cancel := ctxCancelFunc.m[p.Name] - // TODO: delete policy from conf and remove cancelfunc from map. ctxCancelFunc.Unlock() cancel() + if err := deletePolicy(conf, *p); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } natsEncConn.Publish(reply, "policy_delete_ack") // acknowledge policy delete }) @@ -148,3 +152,25 @@ func addSystemDataPolicy(c *config.Config) error { } return nil } + +func deletePolicy(c *config.Config, p policy.Policy) error { + defer ctxCancelFunc.Unlock() + ctxCancelFunc.Lock() + + if _, ok := ctxCancelFunc.m[p.Name]; !ok { + return errors.New("policy not found") + } + + log.Printf("deleting the policy %s...", p.Name) + + delete(ctxCancelFunc.m, p.Name) + for i, q := range c.PolicyConfig { + if q.Name == p.Name { + c.PolicyConfig = append(c.PolicyConfig[:i], c.PolicyConfig[i+1:]...) + } + } + if err := c.Save(); err != nil { + return err + } + return nil +} From 429d2f37a72628410d61cd21e86bd7d742f85e0a Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 2 Sep 2015 12:16:24 +0530 Subject: [PATCH 43/59] move NATS subscription handlers into separate functions --- cmd/recond/main.go | 40 ++------------------------ cmd/recond/subscriptions.go | 56 +++++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 38 deletions(-) create mode 100644 cmd/recond/subscriptions.go diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 162ffa1..3d63724 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -63,44 +63,8 @@ func main() { go runStoredPolicies(conf) - natsEncConn.Subscribe(agent.UID+"_policy_add", func(subj, reply string, p *policy.Policy) { - log.Printf("policy_add 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, "policy_add_ack") // acknowledge policy add - for e := range events { - natsEncConn.Publish("policy_events", e) - } - }) - - natsEncConn.Subscribe(agent.UID+"_policy_delete", func(subj, reply string, p *policy.Policy) { - log.Printf("policy_delete received: %s\n", p.Name) - ctxCancelFunc.Lock() - cancel := ctxCancelFunc.m[p.Name] - ctxCancelFunc.Unlock() - cancel() - if err := deletePolicy(conf, *p); err != nil { - natsEncConn.Publish(reply, err.Error()) - return - } - natsEncConn.Publish(reply, "policy_delete_ack") // acknowledge policy delete - }) + natsEncConn.Subscribe(agent.UID+"_policy_add", AddPolicyHandler(conf)) + natsEncConn.Subscribe(agent.UID+"_policy_delete", DeletePolicyHandler(conf)) // this is just to block the main function from exiting c := make(chan struct{}) diff --git a/cmd/recond/subscriptions.go b/cmd/recond/subscriptions.go new file mode 100644 index 0000000..c2f0672 --- /dev/null +++ b/cmd/recond/subscriptions.go @@ -0,0 +1,56 @@ +// 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("policy_add 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, "policy_add_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("policy_delete received: %s\n", p.Name) + ctxCancelFunc.Lock() + cancel := ctxCancelFunc.m[p.Name] + ctxCancelFunc.Unlock() + cancel() + if err := deletePolicy(conf, *p); err != nil { + natsEncConn.Publish(reply, err.Error()) + return + } + natsEncConn.Publish(reply, "policy_delete_ack") // acknowledge policy delete + } +} From 9fe574b798ca94d0c900e0e6c6924ce540a7a4e4 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 2 Sep 2015 16:31:14 +0530 Subject: [PATCH 44/59] config: add tests for race conditions in (*Config)AddPolicy --- cmd/recond/config/config_test.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/cmd/recond/config/config_test.go b/cmd/recond/config/config_test.go index 6509cf9..489ce11 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) { @@ -131,3 +132,22 @@ 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 + go func() { + if err := c.AddPolicy(p); err != nil { + t.Error(err) + } + }() + go func() { + if err := c.AddPolicy(p); err != nil { + t.Error(err) + } + }() +} From d45cab347c63482f64d0a7aacdb5d824ac3925d4 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 2 Sep 2015 16:51:11 +0530 Subject: [PATCH 45/59] config: fix race condition and tests --- cmd/recond/config/config.go | 6 ++++++ cmd/recond/config/config_test.go | 9 +++------ 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index c3be205..2230de4 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -17,6 +17,7 @@ import ( "os" "os/user" "path/filepath" + "sync" "github.com/codeignition/recon/internal/fileutil" "github.com/codeignition/recon/policy" @@ -38,6 +39,7 @@ func init() { // Config represents the configuration for the recond // running on a particular machine. type Config struct { + sync.Mutex UID string // Unique Identifier to register with marksman PolicyConfig policy.Config } @@ -67,6 +69,8 @@ func Init() (*Config, error) { // Save saves the config in a configuration file. // If it already exists, it removes it and writes it freshly. func (c *Config) Save() error { + defer c.Unlock() + c.Lock() if fileutil.Exists(configPath) { if err := os.Remove(configPath); err != nil { return err @@ -98,6 +102,8 @@ func (c *Config) Save() error { } 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") diff --git a/cmd/recond/config/config_test.go b/cmd/recond/config/config_test.go index 489ce11..548e9dc 100644 --- a/cmd/recond/config/config_test.go +++ b/cmd/recond/config/config_test.go @@ -140,14 +140,11 @@ func TestConcurrentAddPolicy(t *testing.T) { 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() { - if err := c.AddPolicy(p); err != nil { - t.Error(err) - } + c.AddPolicy(p) }() go func() { - if err := c.AddPolicy(p); err != nil { - t.Error(err) - } + c.AddPolicy(p) }() } From 26437c4c8b37e0f7dea973e65154527cd36c1c9e Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 2 Sep 2015 17:02:21 +0530 Subject: [PATCH 46/59] safely delete policies --- cmd/recond/main.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 3d63724..db32d31 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -128,6 +128,8 @@ func deletePolicy(c *config.Config, p policy.Policy) error { log.Printf("deleting the policy %s...", p.Name) delete(ctxCancelFunc.m, p.Name) + defer c.Unlock() + c.Lock() for i, q := range c.PolicyConfig { if q.Name == p.Name { c.PolicyConfig = append(c.PolicyConfig[:i], c.PolicyConfig[i+1:]...) From c3192b030fc7242670dcd08877d109edc0fd3f75 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 3 Sep 2015 11:29:40 +0530 Subject: [PATCH 47/59] remove outdated TODO stub --- cmd/recond/config/config.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 2230de4..27ef07c 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -110,8 +110,6 @@ func (c *Config) AddPolicy(p policy.Policy) error { } } c.PolicyConfig = append(c.PolicyConfig, p) - // TODO: possible race condition. Use a mutex lock while - // writing to the slice PolicyConfig. return nil } From 1133d4dd2bea386b2d7e7d6597d8b0554789744b Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 3 Sep 2015 13:10:10 +0530 Subject: [PATCH 48/59] config: don't lock and unlock the config inside Save If the function calling Save holds the lock, Save waits for the lock to be opened indefinitely. To avoid this behaviour, leave the locking and unlocking upto the caller. --- cmd/recond/config/config.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 27ef07c..39e05ea 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -68,9 +68,8 @@ 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 { - defer c.Unlock() - c.Lock() if fileutil.Exists(configPath) { if err := os.Remove(configPath); err != nil { return err From 27eedd4853fd22fe4f31bb042180efdd21cbfc59 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 3 Sep 2015 13:12:06 +0530 Subject: [PATCH 49/59] add ModifyPolicyHandler --- cmd/recond/main.go | 13 ++++++----- cmd/recond/subscriptions.go | 44 ++++++++++++++++++++++++++++++++++++- 2 files changed, 50 insertions(+), 7 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index db32d31..522029b 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -65,6 +65,7 @@ func main() { natsEncConn.Subscribe(agent.UID+"_policy_add", AddPolicyHandler(conf)) natsEncConn.Subscribe(agent.UID+"_policy_delete", DeletePolicyHandler(conf)) + natsEncConn.Subscribe(agent.UID+"_modify_policy", ModifyPolicyHandler(conf)) // this is just to block the main function from exiting c := make(chan struct{}) @@ -117,24 +118,24 @@ func addSystemDataPolicy(c *config.Config) error { return nil } -func deletePolicy(c *config.Config, p policy.Policy) error { +func deletePolicy(c *config.Config, policyName string) error { defer ctxCancelFunc.Unlock() ctxCancelFunc.Lock() - if _, ok := ctxCancelFunc.m[p.Name]; !ok { + if _, ok := ctxCancelFunc.m[policyName]; !ok { return errors.New("policy not found") } + log.Printf("deleting the policy %s...", policyName) - log.Printf("deleting the policy %s...", p.Name) - - delete(ctxCancelFunc.m, p.Name) + delete(ctxCancelFunc.m, policyName) defer c.Unlock() c.Lock() for i, q := range c.PolicyConfig { - if q.Name == p.Name { + if q.Name == policyName { c.PolicyConfig = append(c.PolicyConfig[:i], c.PolicyConfig[i+1:]...) } } + if err := c.Save(); err != nil { return err } diff --git a/cmd/recond/subscriptions.go b/cmd/recond/subscriptions.go index c2f0672..b10c0ad 100644 --- a/cmd/recond/subscriptions.go +++ b/cmd/recond/subscriptions.go @@ -47,10 +47,52 @@ func DeletePolicyHandler(conf *config.Config) func(subj, reply string, p *policy cancel := ctxCancelFunc.m[p.Name] ctxCancelFunc.Unlock() cancel() - if err := deletePolicy(conf, *p); err != nil { + if err := deletePolicy(conf, p.Name); err != nil { natsEncConn.Publish(reply, err.Error()) return } natsEncConn.Publish(reply, "policy_delete_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) + } + } +} From c35593bc00540107b16a1b6c404f351ca1ea3799 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 3 Sep 2015 13:20:23 +0530 Subject: [PATCH 50/59] change nats channel names and acks to be consistent --- cmd/recond/main.go | 4 ++-- cmd/recond/subscriptions.go | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 522029b..e439528 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -63,8 +63,8 @@ func main() { go runStoredPolicies(conf) - natsEncConn.Subscribe(agent.UID+"_policy_add", AddPolicyHandler(conf)) - natsEncConn.Subscribe(agent.UID+"_policy_delete", DeletePolicyHandler(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 diff --git a/cmd/recond/subscriptions.go b/cmd/recond/subscriptions.go index b10c0ad..c899781 100644 --- a/cmd/recond/subscriptions.go +++ b/cmd/recond/subscriptions.go @@ -14,7 +14,7 @@ import ( func AddPolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Policy) { return func(subj, reply string, p *policy.Policy) { - log.Printf("policy_add received: %s\n", p.Name) + log.Printf("add_policy received: %s\n", p.Name) if err := conf.AddPolicy(*p); err != nil { natsEncConn.Publish(reply, err.Error()) return @@ -33,7 +33,7 @@ func AddPolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Po ctxCancelFunc.m[p.Name] = cancel ctxCancelFunc.Unlock() - natsEncConn.Publish(reply, "policy_add_ack") // acknowledge policy add + natsEncConn.Publish(reply, "add_policy_ack") // acknowledge policy add for e := range events { natsEncConn.Publish("policy_events", e) } @@ -42,7 +42,7 @@ func AddPolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Po func DeletePolicyHandler(conf *config.Config) func(subj, reply string, p *policy.Policy) { return func(subj, reply string, p *policy.Policy) { - log.Printf("policy_delete received: %s\n", p.Name) + log.Printf("delete_policy received: %s\n", p.Name) ctxCancelFunc.Lock() cancel := ctxCancelFunc.m[p.Name] ctxCancelFunc.Unlock() @@ -51,7 +51,7 @@ func DeletePolicyHandler(conf *config.Config) func(subj, reply string, p *policy natsEncConn.Publish(reply, err.Error()) return } - natsEncConn.Publish(reply, "policy_delete_ack") // acknowledge policy delete + natsEncConn.Publish(reply, "delete_policy_ack") // acknowledge policy delete } } From b94aaa868efba19d5123c3299aeb399201e03d5f Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 3 Sep 2015 14:44:44 +0530 Subject: [PATCH 51/59] Add HostName field to Agent --- agent.go | 3 ++- cmd/recond/config/config.go | 8 +++++++- cmd/recond/main.go | 3 ++- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/agent.go b/agent.go index ba1797b..5744991 100644 --- a/agent.go +++ b/agent.go @@ -7,5 +7,6 @@ package recon // Agent represents a recon daemon running on // a machine. type Agent struct { - UID string `json:"uid"` + UID string `json:"uid"` + HostName string `json:"host_name"` } diff --git a/cmd/recond/config/config.go b/cmd/recond/config/config.go index 39e05ea..cebdc20 100644 --- a/cmd/recond/config/config.go +++ b/cmd/recond/config/config.go @@ -41,6 +41,7 @@ func init() { type Config struct { sync.Mutex UID string // Unique Identifier to register with marksman + HostName string PolicyConfig policy.Config } @@ -59,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 diff --git a/cmd/recond/main.go b/cmd/recond/main.go index e439528..4272ce6 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -47,7 +47,8 @@ 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) From b8708b662ebe894765f31156f1ded01ce783c931 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 14 Oct 2015 11:54:26 +0530 Subject: [PATCH 52/59] Accumulate just memory data for now, in system_data policy handler --- policy/handlers/systemdata.go | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/policy/handlers/systemdata.go b/policy/handlers/systemdata.go index 09745f3..f9ae264 100644 --- a/policy/handlers/systemdata.go +++ b/policy/handlers/systemdata.go @@ -10,8 +10,7 @@ import ( "os/user" "time" - "github.com/codeignition/recon/metrics/netstat" - "github.com/codeignition/recon/metrics/ps" + "github.com/codeignition/recon/metrics/memory" "github.com/codeignition/recon/policy" "golang.org/x/net/context" ) @@ -56,19 +55,14 @@ func accumulateSystemData() map[string]interface{} { if err != nil { log.Println(err) } - psdata, err := ps.CollectData() - if err != nil { - log.Println(err) - } - nsdata, err := netstat.CollectData() + memdata, err := memory.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, + "recon_time": time.Now(), + "current_user": currentUser.Username, + "memory": memdata, } return data } From 82d0523a99983b8deb6b9cf9d63b18da4021fa1c Mon Sep 17 00:00:00 2001 From: Hari haran Date: Wed, 14 Oct 2015 12:04:10 +0530 Subject: [PATCH 53/59] Let NATS address be overridden using a command line flag It is useful for testing, when the NATS address sent by marksman references localhost. --- cmd/recond/agent.go | 11 ++++++++++- cmd/recond/main.go | 8 ++++++-- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/cmd/recond/agent.go b/cmd/recond/agent.go index ebe9c7c..085ec61 100644 --- a/cmd/recond/agent.go +++ b/cmd/recond/agent.go @@ -48,7 +48,16 @@ func (a *Agent) register(addr string) error { if err := dec.Decode(&t); err != nil { return err } - nc, err := nats.Connect(t.NatsURL) + + // 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 + } + + nc, err := nats.Connect(naddr) if err != nil { return err } diff --git a/cmd/recond/main.go b/cmd/recond/main.go index 4272ce6..2a48ef8 100644 --- a/cmd/recond/main.go +++ b/cmd/recond/main.go @@ -33,10 +33,14 @@ var ctxCancelFunc = struct { 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() @@ -51,7 +55,7 @@ func main() { HostName: conf.HostName, } - err = agent.register(*marksmanAddr) + err = agent.register(*flagMarksmanAddr) if err != nil { log.Fatalln(err) } From b14c18a72102a35f1850ebd1923d5bd90d5476a8 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 15 Oct 2015 14:10:18 +0530 Subject: [PATCH 54/59] Add top package summarizing system data --- metrics/top/top.go | 203 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 203 insertions(+) create mode 100644 metrics/top/top.go diff --git a/metrics/top/top.go b/metrics/top/top.go new file mode 100644 index 0000000..b31d613 --- /dev/null +++ b/metrics/top/top.go @@ -0,0 +1,203 @@ +// 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" + "time" +) + +// Data represents the `top` data +type Data struct { + Time time.Time + Uptime string + CPU CPUData + Memory MemoryData + Swap SwapData + LoadAvg LoadAverageData +} + +// CPU represents the percentage CPU usages, averaged over the cores +type CPUData struct { + Userspace, + Idle, + System, + IOWait, + Stolen float64 +} + +// LoadAverageData for the last 1min, 5mins, 15mins +type LoadAverageData struct { + LastOneMin, + LastFiveMin, + LastFifteenMin float64 +} + +// MemoryData represents the memory data in kilobytes +type MemoryData struct { + Total, + Used, + Free, + Buffers, + Cached int +} + +// SwapData represents the swap memory data in kilobytes +type SwapData struct { + Total, + Used, + Free int +} + +// CollectData collects the data and returns +// an error if any. +func CollectData() (*Data, error) { + d := new(Data) + out, err := exec.Command("top", "-bn", "1").Output() + if err != nil { + return nil, err + } + d.Time = time.Now() + lines := strings.Split(string(out), "\n") // use a bufio.Scanner if memory problems arise + if len(lines) < 7 { + return nil, errors.New("top: unexpected output") + } + for i, line := range lines { + switch i { + case 0: + err := d.parseUptimeLoadAvgData(line) + if err != nil { + return nil, err + } + case 1: + // we are not collecting the tasks data for now + continue + case 2: + err := d.parseCPUData(line) + if err != nil { + return nil, err + } + case 3: + err := d.parseMemoryData(line) + if err != nil { + return nil, err + } + case 4: + err := d.parseSwapData(line) + if err != nil { + return nil, 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.LoadAvg = LoadAverageData{ + LastOneMin: f[0], + LastFiveMin: f[1], + LastFifteenMin: 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 = CPUData{ + Userspace: c[0], + Idle: c[3], + System: c[1], + IOWait: c[4], + Stolen: c[7], + } + return nil +} + +func (d *Data) parseMemoryData(s string) error { + if strings.HasPrefix(s, "KiB Mem:") { + var m MemoryData + _, err := fmt.Sscanf(s, "KiB Mem:\t%d total,\t%d used,\t%d free,\t%d buffers", &m.Total, &m.Used, &m.Free, &m.Buffers) + if err != nil { + return fmt.Errorf("top: unable to parse memory data; %s", err) + } + d.Memory = m + return nil + } else if strings.HasPrefix(s, "Mem:") { + var m MemoryData + _, err := fmt.Sscanf(s, "Mem:\t%dk total,\t%dk used,\t%dk free,\t%dk buffers", &m.Total, &m.Used, &m.Free, &m.Buffers) + if err != nil { + return fmt.Errorf("top: unable to parse memory data; %s", err) + } + d.Memory = m + return nil + } + return errors.New("top: unable to parse memory data") +} + +func (d *Data) parseSwapData(s string) error { + if strings.HasPrefix(s, "KiB Swap:") { + var sd SwapData + _, err := fmt.Sscanf(s, "KiB Swap:\t%d total,\t%d used,\t%d free.\t%d cached Mem", &sd.Total, &sd.Used, &sd.Free, &d.Memory.Cached) + if err != nil { + return fmt.Errorf("top: unable to parse swap data; %s", err) + } + d.Swap = sd + return nil + } else if strings.HasPrefix(s, "Swap:") { + var sd SwapData + _, err := fmt.Sscanf(s, "Swap:\t%dk total,\t%dk used,\t%dk free,\t%dk cached", &sd.Total, &sd.Used, &sd.Free, &d.Memory.Cached) + if err != nil { + return fmt.Errorf("top: unable to parse swap data; %s", err) + } + d.Swap = sd + return nil + } + return errors.New("top: unable to parse swap data") +} From cb3c4850ebe99eee538102eb2c4be7176a757811 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 15 Oct 2015 16:03:03 +0530 Subject: [PATCH 55/59] Change the Event to not include complete Policy As the number of events sent will be high, having a complete Policy in the Event is pointless. Just keep PolicyName and AgentUID as they are the only useful info --- policy/event.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/policy/event.go b/policy/event.go index 3d94671..e4ee246 100644 --- a/policy/event.go +++ b/policy/event.go @@ -8,7 +8,8 @@ import "time" // Event data that will be sent by policy handlers type Event struct { - Time time.Time - Policy Policy - Data map[string]interface{} // Data may include status, stats, etc. + Time time.Time + PolicyName string `bson:"policy_name"` + AgentUID string `bson:"agent_uid"` + Data interface{} // Data may include status, stats, etc. } From 2840e21fba2107bec50ae74d2e4e427abb01e7cd Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 15 Oct 2015 16:04:50 +0530 Subject: [PATCH 56/59] Change handlers to obey the new Event definition Also, use top.CollectData for systemdata --- policy/handlers/systemdata.go | 27 +++++++++------------------ policy/handlers/tcp.go | 10 ++++++---- 2 files changed, 15 insertions(+), 22 deletions(-) diff --git a/policy/handlers/systemdata.go b/policy/handlers/systemdata.go index f9ae264..353e61c 100644 --- a/policy/handlers/systemdata.go +++ b/policy/handlers/systemdata.go @@ -7,10 +7,9 @@ package handlers import ( "errors" "log" - "os/user" "time" - "github.com/codeignition/recon/metrics/memory" + "github.com/codeignition/recon/metrics/top" "github.com/codeignition/recon/policy" "golang.org/x/net/context" ) @@ -40,9 +39,10 @@ func SystemData(ctx context.Context, p policy.Policy) (<-chan policy.Event, erro return case <-t.C: out <- policy.Event{ - Time: time.Now(), - Policy: p, - Data: accumulateSystemData(), + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, + Data: accumulateSystemData(), } } } @@ -50,19 +50,10 @@ func SystemData(ctx context.Context, p policy.Policy) (<-chan policy.Event, erro return out, nil } -func accumulateSystemData() map[string]interface{} { - currentUser, err := user.Current() +func accumulateSystemData() interface{} { + d, err := top.CollectData() if err != nil { - log.Println(err) + log.Print(err) } - memdata, err := memory.CollectData() - if err != nil { - log.Println(err) - } - data := map[string]interface{}{ - "recon_time": time.Now(), - "current_user": currentUser.Username, - "memory": memdata, - } - return data + return d } diff --git a/policy/handlers/tcp.go b/policy/handlers/tcp.go index cb151d3..3974abb 100644 --- a/policy/handlers/tcp.go +++ b/policy/handlers/tcp.go @@ -55,8 +55,9 @@ func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { _, err := net.DialTimeout("tcp", addr, d) if err != nil { out <- policy.Event{ - Time: time.Now(), - Policy: p, + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, Data: map[string]interface{}{ "status": "failure", "error": err.Error(), @@ -64,8 +65,9 @@ func TCP(ctx context.Context, p policy.Policy) (<-chan policy.Event, error) { } } else { out <- policy.Event{ - Time: time.Now(), - Policy: p, + Time: time.Now(), + PolicyName: p.Name, + AgentUID: p.AgentUID, Data: map[string]interface{}{ "status": "success", }, From 838a48bcb8bc1dd6095fc34c4581202c9aa7b528 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 15 Oct 2015 16:17:23 +0530 Subject: [PATCH 57/59] top: remove Time field from Data Time field is already present in the Event field, Adding the time field in top.Data is redundant. The difference of milliseconds while collecting the data won't make much difference. --- metrics/top/top.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/metrics/top/top.go b/metrics/top/top.go index b31d613..ae3ae32 100644 --- a/metrics/top/top.go +++ b/metrics/top/top.go @@ -14,12 +14,10 @@ import ( "os/exec" "strconv" "strings" - "time" ) // Data represents the `top` data type Data struct { - Time time.Time Uptime string CPU CPUData Memory MemoryData @@ -67,7 +65,6 @@ func CollectData() (*Data, error) { if err != nil { return nil, err } - d.Time = time.Now() lines := strings.Split(string(out), "\n") // use a bufio.Scanner if memory problems arise if len(lines) < 7 { return nil, errors.New("top: unexpected output") From 8ebfb0c6f1c26f137787bdbbb5d3a3c316d39303 Mon Sep 17 00:00:00 2001 From: Hari haran Date: Tue, 27 Oct 2015 13:17:16 +0530 Subject: [PATCH 58/59] top: run 2 iterations of top in batch mode The 1st iteration gives bad results, so we ignore the output of the 1st iteration and only consider the 2nd iteration's output. --- metrics/top/top.go | 61 ++++++++++++++++++++++++++-------------------- 1 file changed, 35 insertions(+), 26 deletions(-) diff --git a/metrics/top/top.go b/metrics/top/top.go index ae3ae32..68538c6 100644 --- a/metrics/top/top.go +++ b/metrics/top/top.go @@ -61,7 +61,9 @@ type SwapData struct { // an error if any. func CollectData() (*Data, error) { d := new(Data) - out, err := exec.Command("top", "-bn", "1").Output() + 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 nil, err } @@ -69,33 +71,40 @@ func CollectData() (*Data, error) { if len(lines) < 7 { return nil, errors.New("top: unexpected output") } + base := 0 for i, line := range lines { - switch i { - case 0: - err := d.parseUptimeLoadAvgData(line) - if err != nil { - return nil, err - } - case 1: - // we are not collecting the tasks data for now - continue - case 2: - err := d.parseCPUData(line) - if err != nil { - return nil, err - } - case 3: - err := d.parseMemoryData(line) - if err != nil { - return nil, err - } - case 4: - err := d.parseSwapData(line) - if err != nil { - return nil, err + if strings.HasPrefix(line, "top") { + count++ + base = i + } + if count == iters { + switch i - base { + case 0: + err := d.parseUptimeLoadAvgData(line) + if err != nil { + return nil, err + } + case 1: + // we are not collecting the tasks data for now + continue + case 2: + err := d.parseCPUData(line) + if err != nil { + return nil, err + } + case 3: + err := d.parseMemoryData(line) + if err != nil { + return nil, err + } + case 4: + err := d.parseSwapData(line) + if err != nil { + return nil, err + } + default: + break } - default: - break } } return d, nil From 6865f782ec74794035aa893af5b6d16cd3d5f06e Mon Sep 17 00:00:00 2001 From: Hari haran Date: Thu, 29 Oct 2015 11:28:57 +0530 Subject: [PATCH 59/59] Add metrics/system package Also move all the unwanted packages into misc/ Use system package for system data policy. --- .../blockdevice/blockdevice_linux.go | 0 metrics/{ => misc}/counters/counters.go | 0 metrics/{ => misc}/cpu/cpu_linux.go | 0 metrics/{ => misc}/etc/etc_linux.go | 0 metrics/{ => misc}/filesystem/filesystem.go | 0 .../initpackage/initpackage_linux.go | 0 metrics/{ => misc}/kernel/kernel_linux.go | 0 metrics/{ => misc}/languages/languages.go | 0 metrics/{ => misc}/lsb/lsb_linux.go | 0 metrics/{ => misc}/memory/memory_linux.go | 0 metrics/{ => misc}/netstat/netstat.go | 0 metrics/{ => misc}/network/network.go | 0 metrics/{ => misc}/ps/ps.go | 0 metrics/{ => misc}/uptime/uptime_linux.go | 0 metrics/system/disk/disk.go | 52 +++++++ metrics/system/system.go | 40 +++++ metrics/{ => system}/top/top.go | 146 +++++++----------- policy/handlers/systemdata.go | 9 +- 18 files changed, 151 insertions(+), 96 deletions(-) rename metrics/{ => misc}/blockdevice/blockdevice_linux.go (100%) rename metrics/{ => misc}/counters/counters.go (100%) rename metrics/{ => misc}/cpu/cpu_linux.go (100%) rename metrics/{ => misc}/etc/etc_linux.go (100%) rename metrics/{ => misc}/filesystem/filesystem.go (100%) rename metrics/{ => misc}/initpackage/initpackage_linux.go (100%) rename metrics/{ => misc}/kernel/kernel_linux.go (100%) rename metrics/{ => misc}/languages/languages.go (100%) rename metrics/{ => misc}/lsb/lsb_linux.go (100%) rename metrics/{ => misc}/memory/memory_linux.go (100%) rename metrics/{ => misc}/netstat/netstat.go (100%) rename metrics/{ => misc}/network/network.go (100%) rename metrics/{ => misc}/ps/ps.go (100%) rename metrics/{ => misc}/uptime/uptime_linux.go (100%) create mode 100644 metrics/system/disk/disk.go create mode 100644 metrics/system/system.go rename metrics/{ => system}/top/top.go (50%) 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/top/top.go b/metrics/system/top/top.go similarity index 50% rename from metrics/top/top.go rename to metrics/system/top/top.go index 68538c6..b4ef26c 100644 --- a/metrics/top/top.go +++ b/metrics/system/top/top.go @@ -16,60 +16,21 @@ import ( "strings" ) -// Data represents the `top` data -type Data struct { - Uptime string - CPU CPUData - Memory MemoryData - Swap SwapData - LoadAvg LoadAverageData -} - -// CPU represents the percentage CPU usages, averaged over the cores -type CPUData struct { - Userspace, - Idle, - System, - IOWait, - Stolen float64 -} - -// LoadAverageData for the last 1min, 5mins, 15mins -type LoadAverageData struct { - LastOneMin, - LastFiveMin, - LastFifteenMin float64 -} - -// MemoryData represents the memory data in kilobytes -type MemoryData struct { - Total, - Used, - Free, - Buffers, - Cached int -} - -// SwapData represents the swap memory data in kilobytes -type SwapData struct { - Total, - Used, - Free int -} +type Data map[string]interface{} // CollectData collects the data and returns // an error if any. -func CollectData() (*Data, error) { - d := new(Data) +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 nil, err + return d, err } lines := strings.Split(string(out), "\n") // use a bufio.Scanner if memory problems arise if len(lines) < 7 { - return nil, errors.New("top: unexpected output") + return d, errors.New("top: unexpected output") } base := 0 for i, line := range lines { @@ -82,7 +43,7 @@ func CollectData() (*Data, error) { case 0: err := d.parseUptimeLoadAvgData(line) if err != nil { - return nil, err + return d, err } case 1: // we are not collecting the tasks data for now @@ -90,17 +51,17 @@ func CollectData() (*Data, error) { case 2: err := d.parseCPUData(line) if err != nil { - return nil, err + return d, err } case 3: err := d.parseMemoryData(line) if err != nil { - return nil, err + return d, err } case 4: err := d.parseSwapData(line) if err != nil { - return nil, err + return d, err } default: break @@ -110,7 +71,7 @@ func CollectData() (*Data, error) { return d, nil } -func (d *Data) parseUptimeLoadAvgData(s string) error { +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) @@ -119,7 +80,7 @@ func (d *Data) parseUptimeLoadAvgData(s string) error { if i == -1 { return errors.New("top: unable to parse uptime data") } - d.Uptime = c[:i] + d["uptime"] = c[:i] l := strings.Split(a[1], ",") if len(l) != 3 { return errors.New("top: unexpected number of load averages found") @@ -132,15 +93,15 @@ func (d *Data) parseUptimeLoadAvgData(s string) error { return err } } - d.LoadAvg = LoadAverageData{ - LastOneMin: f[0], - LastFiveMin: f[1], - LastFifteenMin: f[2], + 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 { +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, ",") @@ -156,54 +117,53 @@ func (d *Data) parseCPUData(s string) error { return err } } - d.CPU = CPUData{ - Userspace: c[0], - Idle: c[3], - System: c[1], - IOWait: c[4], - Stolen: c[7], + 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 { +func (d Data) parseMemoryData(s string) error { + var total, used, free, buffers int + var err error if strings.HasPrefix(s, "KiB Mem:") { - var m MemoryData - _, err := fmt.Sscanf(s, "KiB Mem:\t%d total,\t%d used,\t%d free,\t%d buffers", &m.Total, &m.Used, &m.Free, &m.Buffers) - if err != nil { - return fmt.Errorf("top: unable to parse memory data; %s", err) - } - d.Memory = m - return nil + _, 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:") { - var m MemoryData - _, err := fmt.Sscanf(s, "Mem:\t%dk total,\t%dk used,\t%dk free,\t%dk buffers", &m.Total, &m.Used, &m.Free, &m.Buffers) - if err != nil { - return fmt.Errorf("top: unable to parse memory data; %s", err) - } - d.Memory = m - return nil + _, err = fmt.Sscanf(s, "Mem:\t%dk total,\t%dk used,\t%dk free,\t%dk buffers", &total, &used, &free, &buffers) } - return errors.New("top: unable to parse memory data") + 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 { +func (d Data) parseSwapData(s string) error { + var total, used, free, cached int + var err error if strings.HasPrefix(s, "KiB Swap:") { - var sd SwapData - _, err := fmt.Sscanf(s, "KiB Swap:\t%d total,\t%d used,\t%d free.\t%d cached Mem", &sd.Total, &sd.Used, &sd.Free, &d.Memory.Cached) - if err != nil { - return fmt.Errorf("top: unable to parse swap data; %s", err) - } - d.Swap = sd - return nil + _, 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:") { - var sd SwapData - _, err := fmt.Sscanf(s, "Swap:\t%dk total,\t%dk used,\t%dk free,\t%dk cached", &sd.Total, &sd.Used, &sd.Free, &d.Memory.Cached) - if err != nil { - return fmt.Errorf("top: unable to parse swap data; %s", err) - } - d.Swap = sd - return nil + _, 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, } - return errors.New("top: unable to parse swap data") + mem := d["memory"].(Data) + mem["cached"] = cached + return nil } diff --git a/policy/handlers/systemdata.go b/policy/handlers/systemdata.go index 353e61c..90bd0da 100644 --- a/policy/handlers/systemdata.go +++ b/policy/handlers/systemdata.go @@ -9,7 +9,7 @@ import ( "log" "time" - "github.com/codeignition/recon/metrics/top" + "github.com/codeignition/recon/metrics/system" "github.com/codeignition/recon/policy" "golang.org/x/net/context" ) @@ -51,9 +51,12 @@ func SystemData(ctx context.Context, p policy.Policy) (<-chan policy.Event, erro } func accumulateSystemData() interface{} { - d, err := top.CollectData() + d, err := system.CollectData() if err != nil { log.Print(err) } - return d + a := map[string]interface{}{ + "system": d, + } + return a }