Major code refactor

Author: rs

README.md

@@ -9,7 +9,7 @@ Above capture is jplot monitoring a Go service's [expvar](https://golang.org/pkg
 
 ```
 jplot --url http://:8080/debug/vars \
-    memstats.HeapAlloc+memstats.HeapSys+memstats.HeapAlloc+memstats.HeapIdle+marker:counter:memstats.NumGC \
+    memstats.HeapAlloc+memstats.HeapSys+memstats.HeapAlloc+memstats.HeapIdle+marker,counter:memstats.NumGC \
     counter:memstats.TotalAlloc \
     memstats.HeapObjects \
     memstats.StackSys+memstats.StackInuse
@@ -85,7 +85,7 @@ Here is an example command to graph a Go program memstats:
 
 ```
 jplot --url http://:8080/debug/vars \
-    memstats.HeapAlloc+memstats.HeapSys+memstats.HeapAlloc+memstats.HeapIdle+marker:counter:memstats.NumGC \
+    memstats.HeapAlloc+memstats.HeapSys+memstats.HeapAlloc+memstats.HeapIdle+marker,counter:memstats.NumGC \
     counter:memstats.TotalAlloc \
     memstats.HeapObjects \
     memstats.StackSys+memstats.StackInuse

data/datapoint.go

@@ -1,4 +1,4 @@
-package source
+package data
 
 import (
 	"io/ioutil"
@@ -8,7 +8,7 @@ import (
 	"github.com/elgs/gojq"
 )
 
-type HTTP struct {
+type httpSource struct {
 	c    chan res
 	done chan struct{}
 }
@@ -18,16 +18,20 @@ type res struct {
 	err error
 }
 
-func NewHTTP(url string, interval time.Duration) HTTP {
-	h := HTTP{
+// FromHTTP fetch data points from url every interval and keep size points.
+func FromHTTP(url string, interval time.Duration, size int) *Points {
+	h := httpSource{
 		c:    make(chan res),
 		done: make(chan struct{}),
 	}
 	go h.run(url, interval)
-	return h
+	return &Points{
+		Size:   size,
+		Source: h,
+	}
 }
 
-func (h HTTP) run(url string, interval time.Duration) {
+func (h httpSource) run(url string, interval time.Duration) {
 	t := time.NewTicker(interval)
 	defer t.Stop()
 	h.fetch(url)
@@ -41,7 +45,7 @@ func (h HTTP) run(url string, interval time.Duration) {
 	}
 }
 
-func (h HTTP) fetch(url string) {
+func (h httpSource) fetch(url string) {
 	resp, err := http.Get(url)
 	if err != nil {
 		h.c <- res{err: err}
@@ -57,12 +61,12 @@ func (h HTTP) fetch(url string) {
 	h.c <- res{jq: jq, err: err}
 }
 
-func (h HTTP) Get() (*gojq.JQ, error) {
+func (h httpSource) Get() (*gojq.JQ, error) {
 	res := <-h.c
 	return res.jq, res.err
 }
 
-func (h HTTP) Close() error {
+func (h httpSource) Close() error {
 	close(h.done)
 	return nil
 }

source/http.go renamed to data/http.go

@@ -0,0 +1,94 @@
+package data
+
+import (
+	"fmt"
+	"io"
+	"sync"
+
+	"github.com/elgs/gojq"
+)
+
+// Getter is the interface used by Points to get next points.
+type Getter interface {
+	io.Closer
+	Get() (*gojq.JQ, error)
+}
+
+// Points is a series of Size data points gathered from Source.
+type Points struct {
+	// Size is the number of data point to store per metric.
+	Size   int
+	Source Getter
+
+	points map[string][]float64
+	last   map[string]float64
+	mu     sync.Mutex
+}
+
+// Run get data from the source and capture metrics following specs.
+func (p *Points) Run(specs []Spec) error {
+	for {
+		jq, err := p.Source.Get()
+		if err != nil {
+			return fmt.Errorf("input error: %v", err)
+		}
+		if jq == nil {
+			break
+		}
+		for _, spec := range specs {
+			for _, f := range spec.Fields {
+				v, err := jq.Query(f.Name)
+				if err != nil {
+					return fmt.Errorf("cannot get %s: %v", f.Name, err)
+				}
+				n, ok := v.(float64)
+				if !ok {
+					return fmt.Errorf("invalid type %s: %T", f.Name, v)
+				}
+				p.push(f.ID, n, f.IsCounter)
+			}
+		}
+	}
+	return nil
+}
+
+func (p *Points) push(name string, value float64, counter bool) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	d := p.getLocked(name)
+	if counter {
+		var diff float64
+		if last := p.last[name]; last > 0 {
+			diff = value - last
+		}
+		p.last[name] = value
+		value = diff
+	}
+	d = append(append(make([]float64, 0, p.Size), d[1:]...), value)
+	p.points[name] = d
+}
+
+// Get gets the points vector for name.
+func (p *Points) Get(name string) []float64 {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.getLocked(name)
+}
+
+func (p *Points) getLocked(name string) []float64 {
+	if p.points == nil {
+		p.points = make(map[string][]float64, 1)
+		p.last = make(map[string]float64)
+	}
+	d, found := p.points[name]
+	if !found {
+		d = make([]float64, p.Size)
+		p.points[name] = d
+	}
+	return d
+}
+
+// Close calls Close on Source.
+func (p *Points) Close() error {
+	return p.Source.Close()
+}

data/point.go

@@ -0,0 +1,59 @@
+package data
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Spec specify a list of field for a single graph.
+type Spec struct {
+	Fields []Field
+}
+
+// Field describe a field in a graph.
+type Field struct {
+	ID        string
+	Name      string
+	IsCounter bool
+	IsMarker  bool
+}
+
+// ParseSpec parses a graph specification. Each spec is a string with one or
+// more JSON path separated by + with fields options prefixed with colon and
+// separated by commas.
+func ParseSpec(args []string) ([]Spec, error) {
+	specs := make([]Spec, 0, len(args))
+	for i, v := range args {
+		spec := Spec{}
+		for j, name := range strings.Split(v, "+") {
+			var isCounter bool
+			var isMarker bool
+			if strings.HasPrefix(name, "marker:counter:") {
+				// Backward compat.
+				name = strings.Replace(name, "marker:counter:", "marker,counter:", 1)
+			}
+			if idx := strings.IndexByte(name, ':'); idx != -1 {
+				options := strings.Split(name[:idx], ",")
+				name = name[idx+1:]
+				for _, o := range options {
+					switch o {
+					case "counter":
+						isCounter = true
+					case "marker":
+						isMarker = true
+					default:
+						return nil, fmt.Errorf("invalid field option: %s", o)
+					}
+				}
+			}
+			spec.Fields = append(spec.Fields, Field{
+				ID:        fmt.Sprintf("%d.%d.%s", i, j, name),
+				Name:      name,
+				IsCounter: isCounter,
+				IsMarker:  isMarker,
+			})
+		}
+		specs = append(specs, spec)
+	}
+	return specs, nil
+}

data/spec.go

@@ -0,0 +1,31 @@
+package data
+
+import (
+	"bufio"
+	"os"
+
+	"github.com/elgs/gojq"
+)
+
+type stdin struct {
+	scan *bufio.Scanner
+}
+
+// FromStdin reads data from stdin as one JSON object per line.
+func FromStdin(size int) *Points {
+	return &Points{
+		Size:   size,
+		Source: stdin{bufio.NewScanner(os.Stdin)},
+	}
+}
+
+func (s stdin) Get() (*gojq.JQ, error) {
+	if s.scan.Scan() {
+		return gojq.NewStringQuery(s.scan.Text())
+	}
+	return nil, s.scan.Err()
+}
+
+func (s stdin) Close() error {
+	return nil
+}

data/stdin.go

@@ -0,0 +1,35 @@
+package graph
+
+import (
+	"image"
+	"image/draw"
+	"image/png"
+	"io"
+
+	"github.com/rs/jplot/data"
+	chart "github.com/wcharczuk/go-chart"
+)
+
+type Dash struct {
+	Specs []data.Spec
+	Data  *data.Points
+}
+
+// Render generates a PNG with all graphs stacked.
+func (d Dash) Render(w io.Writer, width, height int) {
+	graphs := make([]chart.Chart, 0, len(d.Specs))
+	for _, spec := range d.Specs {
+		graphs = append(graphs, New(spec, d.Data, width, height/len(d.Specs)))
+	}
+	canvas := image.NewRGBA(image.Rectangle{image.Point{0, 0}, image.Point{width, height}})
+	var top int
+	for _, graph := range graphs {
+		iw := &chart.ImageWriter{}
+		graph.Render(chart.PNG, iw)
+		img, _ := iw.Image()
+		r := image.Rectangle{image.Point{0, top}, image.Point{width, top + graph.Height}}
+		top += graph.Height
+		draw.Draw(canvas, r, img, image.Point{0, 0}, draw.Src)
+	}
+	png.Encode(w, canvas)
+}