0.3.0 Removing internal parallelism: httpjson and exec

This commit is contained in:
Cameron Sparr 2015-12-19 15:29:07 -07:00
parent c584129758
commit b0f79f43ec
3 changed files with 62 additions and 154 deletions

View File

@ -3,13 +3,8 @@ package exec
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"math"
"os/exec"
"strings"
"sync"
"time"
"github.com/gonuts/go-shellquote"
@ -18,47 +13,28 @@ import (
)
const sampleConfig = `
# specify commands via an array of tables
[[plugins.exec.commands]]
# the command to run
command = "/usr/bin/mycollector --foo=bar"
# name of the command (used as a prefix for measurements)
name = "mycollector"
# Only run this command if it has been at least this many
# seconds since it last ran
interval = 10
`
type Exec struct {
Commands []*Command
runner Runner
clock Clock
}
Command string
Name string
type Command struct {
Command string
Name string
Interval int
lastRunAt time.Time
runner Runner
}
type Runner interface {
Run(*Command) ([]byte, error)
}
type Clock interface {
Now() time.Time
Run(*Exec) ([]byte, error)
}
type CommandRunner struct{}
type RealClock struct{}
func (c CommandRunner) Run(command *Command) ([]byte, error) {
command.lastRunAt = time.Now()
split_cmd, err := shellquote.Split(command.Command)
func (c CommandRunner) Run(e *Exec) ([]byte, error) {
split_cmd, err := shellquote.Split(e.Command)
if err != nil || len(split_cmd) == 0 {
return nil, fmt.Errorf("exec: unable to parse command, %s", err)
}
@ -68,18 +44,14 @@ func (c CommandRunner) Run(command *Command) ([]byte, error) {
cmd.Stdout = &out
if err := cmd.Run(); err != nil {
return nil, fmt.Errorf("exec: %s for command '%s'", err, command.Command)
return nil, fmt.Errorf("exec: %s for command '%s'", err, e.Command)
}
return out.Bytes(), nil
}
func (c RealClock) Now() time.Time {
return time.Now()
}
func NewExec() *Exec {
return &Exec{runner: CommandRunner{}, clock: RealClock{}}
return &Exec{runner: CommandRunner{}}
}
func (e *Exec) SampleConfig() string {
@ -91,72 +63,31 @@ func (e *Exec) Description() string {
}
func (e *Exec) Gather(acc plugins.Accumulator) error {
var wg sync.WaitGroup
errorChannel := make(chan error, len(e.Commands))
for _, c := range e.Commands {
wg.Add(1)
go func(c *Command, acc plugins.Accumulator) {
defer wg.Done()
err := e.gatherCommand(c, acc)
if err != nil {
errorChannel <- err
}
}(c, acc)
out, err := e.runner.Run(e)
if err != nil {
return err
}
wg.Wait()
close(errorChannel)
// Get all errors and return them as one giant error
errorStrings := []string{}
for err := range errorChannel {
errorStrings = append(errorStrings, err.Error())
var jsonOut interface{}
err = json.Unmarshal(out, &jsonOut)
if err != nil {
return fmt.Errorf("exec: unable to parse output of '%s' as JSON, %s",
e.Command, err)
}
if len(errorStrings) == 0 {
return nil
f := internal.JSONFlattener{}
err = f.FlattenJSON("", jsonOut)
if err != nil {
return err
}
return errors.New(strings.Join(errorStrings, "\n"))
}
func (e *Exec) gatherCommand(c *Command, acc plugins.Accumulator) error {
secondsSinceLastRun := 0.0
if c.lastRunAt.Unix() == 0 { // means time is uninitialized
secondsSinceLastRun = math.Inf(1)
var msrmnt_name string
if e.Name == "" {
msrmnt_name = "exec"
} else {
secondsSinceLastRun = (e.clock.Now().Sub(c.lastRunAt)).Seconds()
}
if secondsSinceLastRun >= float64(c.Interval) {
out, err := e.runner.Run(c)
if err != nil {
return err
}
var jsonOut interface{}
err = json.Unmarshal(out, &jsonOut)
if err != nil {
return fmt.Errorf("exec: unable to parse output of '%s' as JSON, %s",
c.Command, err)
}
f := internal.JSONFlattener{}
err = f.FlattenJSON("", jsonOut)
if err != nil {
return err
}
var msrmnt_name string
if c.Name == "" {
msrmnt_name = "exec"
} else {
msrmnt_name = "exec_" + c.Name
}
acc.AddFields(msrmnt_name, f.Fields, nil)
msrmnt_name = "exec_" + e.Name
}
acc.AddFields(msrmnt_name, f.Fields, nil)
return nil
}

View File

@ -15,16 +15,12 @@ import (
)
type HttpJson struct {
Services []Service
client HTTPClient
}
type Service struct {
Name string
Servers []string
Method string
TagKeys []string
Parameters map[string]string
client HTTPClient
}
type HTTPClient interface {
@ -48,31 +44,28 @@ func (c RealHTTPClient) MakeRequest(req *http.Request) (*http.Response, error) {
}
var sampleConfig = `
# Specify services via an array of tables
[[plugins.httpjson.services]]
# a name for the service being polled
name = "webserver_stats"
# a name for the service being polled
name = "webserver_stats"
# URL of each server in the service's cluster
servers = [
"http://localhost:9999/stats/",
"http://localhost:9998/stats/",
]
# URL of each server in the service's cluster
servers = [
"http://localhost:9999/stats/",
"http://localhost:9998/stats/",
]
# HTTP method to use (case-sensitive)
method = "GET"
# HTTP method to use (case-sensitive)
method = "GET"
# List of tag names to extract from top-level of JSON server response
# tag_keys = [
# "my_tag_1",
# "my_tag_2"
# ]
# List of tag names to extract from top-level of JSON server response
# tag_keys = [
# "my_tag_1",
# "my_tag_2"
# ]
# HTTP parameters (all values must be strings)
[plugins.httpjson.services.parameters]
event_type = "cpu_spike"
threshold = "0.75"
# HTTP parameters (all values must be strings)
[plugins.httpjson.parameters]
event_type = "cpu_spike"
threshold = "0.75"
`
func (h *HttpJson) SampleConfig() string {
@ -87,22 +80,16 @@ func (h *HttpJson) Description() string {
func (h *HttpJson) Gather(acc plugins.Accumulator) error {
var wg sync.WaitGroup
totalServers := 0
for _, service := range h.Services {
totalServers += len(service.Servers)
}
errorChannel := make(chan error, totalServers)
errorChannel := make(chan error, len(h.Servers))
for _, service := range h.Services {
for _, server := range service.Servers {
wg.Add(1)
go func(service Service, server string) {
defer wg.Done()
if err := h.gatherServer(acc, service, server); err != nil {
errorChannel <- err
}
}(service, server)
}
for _, server := range h.Servers {
wg.Add(1)
go func(server string) {
defer wg.Done()
if err := h.gatherServer(acc, server); err != nil {
errorChannel <- err
}
}(server)
}
wg.Wait()
@ -130,10 +117,9 @@ func (h *HttpJson) Gather(acc plugins.Accumulator) error {
// error: Any error that may have occurred
func (h *HttpJson) gatherServer(
acc plugins.Accumulator,
service Service,
serverURL string,
) error {
resp, err := h.sendRequest(service, serverURL)
resp, err := h.sendRequest(serverURL)
if err != nil {
return err
}
@ -147,7 +133,7 @@ func (h *HttpJson) gatherServer(
"server": serverURL,
}
for _, tag := range service.TagKeys {
for _, tag := range h.TagKeys {
switch v := jsonOut[tag].(type) {
case string:
tags[tag] = v
@ -162,10 +148,10 @@ func (h *HttpJson) gatherServer(
}
var msrmnt_name string
if service.Name == "" {
if h.Name == "" {
msrmnt_name = "httpjson"
} else {
msrmnt_name = "httpjson_" + service.Name
msrmnt_name = "httpjson_" + h.Name
}
acc.AddFields(msrmnt_name, f.Fields, nil)
return nil
@ -178,7 +164,7 @@ func (h *HttpJson) gatherServer(
// Returns:
// string: body of the response
// error : Any error that may have occurred
func (h *HttpJson) sendRequest(service Service, serverURL string) (string, error) {
func (h *HttpJson) sendRequest(serverURL string) (string, error) {
// Prepare URL
requestURL, err := url.Parse(serverURL)
if err != nil {
@ -186,21 +172,23 @@ func (h *HttpJson) sendRequest(service Service, serverURL string) (string, error
}
params := url.Values{}
for k, v := range service.Parameters {
for k, v := range h.Parameters {
params.Add(k, v)
}
requestURL.RawQuery = params.Encode()
// Create + send request
req, err := http.NewRequest(service.Method, requestURL.String(), nil)
req, err := http.NewRequest(h.Method, requestURL.String(), nil)
if err != nil {
return "", err
}
defer req.Body.Close()
resp, err := h.client.MakeRequest(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)

View File

@ -62,17 +62,6 @@ func (j *Jolokia) SampleConfig() string {
[[plugins.jolokia.metrics]]
name = "heap_memory_usage"
jmx = "/java.lang:type=Memory/HeapMemoryUsage"
# This drops the 'committed' value from Eden space measurement
[[plugins.jolokia.metrics]]
name = "memory_eden"
jmx = "/java.lang:type=MemoryPool,name=PS Eden Space/Usage"
# This passes only DaemonThreadCount and ThreadCount
[[plugins.jolokia.metrics]]
name = "heap_threads"
jmx = "/java.lang:type=Threading"
`
}