Initial revision

.travis.yml

@@ -0,0 +1,11 @@
+language: go
+go:
+- "1.8"
+- "1.9"
+- "1.10"
+- master
+matrix:
+  allow_failures:
+      - go: master
+script:
+  go test -v -race -cpu=1,2,4 -bench . -benchmem ./...

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Olivier Poitrey
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,60 @@
+# jaggr: JSON Aggregation CLI
+[![license](http://img.shields.io/badge/license-MIT-red.svg?style=flat)](https://raw.githubusercontent.com/rs/jaggr/master/LICENSE) [![Build Status](https://travis-ci.org/rs/jaggr.svg?branch=master)](https://travis-ci.org/rs/jaggr)
+
+Jaggr is a command line tool to aggregate in real time a series of JSON logs. The main goal of this tool is to prepare data for plotting with [jplot](https://github.com/rs/jplot).
+
+## Install
+
+```
+go get -u github.com/rs/jaggr
+```
+
+## Usage
+
+Given the input below, generate one line per second with mean, min, max:
+
+```
+{"code": 200, "latency": 4788000, "error": ""}
+{"code": 200, "latency": 5785000, "error": ""}
+{"code": 200, "latency": 4162000, "error": ""}
+{"code": 502, "latency": 4461000, "error": "i/o error"}
+{"code": 200, "latency": 5884000, "error": ""}
+{"code": 200, "latency": 4702000, "error": ""}
+...
+```
+
+```
+tail -f log.json | jaggr @count=rps hist[200,502,*]:code min,max,mean:latency
+```
+
+Output will be on line per second as follow:
+
+```
+{"count":123, "code": {"hist": {"200": 100, 502: 13, "*": 10}}, "latency":{"min": 4461000, "max": 5884000, "mean": 4483000}}
+```
+
+So here we give a stream of real-time requests to jaggr standard input and request the aggregation of the `code` and `latency` fields. For the `code` we request an histogram with some known error codes with an "other" bucket defined by `*`. The `latency` field is aggregated using minimum, maximum and mean. In addition, `@count` adds an extra field indicating the total number of lines aggregated. The `=` sign can be used on any field to rename it, here we use it to say that the count is an `rps` as we are using the default aggregation time of 1 second.
+
+Note that any field not specified in the argument list are removed from the output (i.e. `error` field).
+
+## Recipes
+
+### Vegeta
+
+Jaggr can be used to integrate [vegeta](https://github.com/tsenart/vegeta) with [jplot](https://github.com/rs/jplot) as follow:
+
+```
+echo 'GET http://localhost' | \
+    vegeta attack -rate 5000 -workers 100 -duration 10m | vegeta dump | \
+    jaggr @count=rps \
+          hist[100,200,300,400,500]:code \
+          p25,p50,p95:latency \
+          p25,p50,p95:bytes_in \
+          p25,p50,p95:bytes_out | \
+    jplot rps+code.hist.100+code.hist.200+code.hist.300+code.hist.400+code.hist.500 \
+          latency.p95+latency.p50+latency.p25 \
+          bytes_in.p95+bytes_in.p50+bytes_in.p25 \
+          bytes_out.p95+bytes_out.p50+bytes_out.p25
+```
+
+![](doc/vegeta.gif)

aggr/aggr.go

@@ -0,0 +1,45 @@
+package aggr
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+)
+
+var rAggrs = map[string]func(args []string) (Aggregator, error){}
+
+func init() {
+	Register("min", newMin)
+	Register("max", newMax)
+	Register("mean", newMean)
+	Register("median", newMedian)
+	Register("hist", newHist)
+	Register("cat", newCat)
+	for i := 1; i < 100; i++ {
+		p := float64(i)
+		Register(fmt.Sprintf("p%d", i), func(args []string) (Aggregator, error) {
+			return &percentile{p: p}, nil
+		})
+	}
+}
+
+// Aggregator accumulates data using Push and aggregate (and reset) the
+// accumulated data on Aggr. The result is expected to be used as a JSON value.
+// Numbers are expressed as float64.
+type Aggregator interface {
+	Push(interface{}) error
+	Aggr() interface{}
+}
+
+// Register a custom aggregator.
+func Register(name string, aggr func(args []string) (Aggregator, error)) error {
+	if _, found := rAggrs[name]; found {
+		return errors.New("already exists")
+	}
+	rAggrs[name] = aggr
+	return nil
+}
+
+func parseFloat(v interface{}) (float64, error) {
+	return strconv.ParseFloat(fmt.Sprint(v), 64)
+}

aggr/cat.go

@@ -0,0 +1,34 @@
+package aggr
+
+import (
+	"fmt"
+)
+
+type cat map[string]uint32
+
+func newCat(args []string) (Aggregator, error) {
+	h := cat{}
+	for _, bucket := range args {
+		h[bucket] = 0
+	}
+	return h, nil
+}
+
+func (a cat) Push(v interface{}) error {
+	bucket := fmt.Sprint(v)
+	if _, found := a[bucket]; found {
+		a[bucket]++
+	} else if _, found := a["*"]; found {
+		a["*"]++
+	}
+	return nil
+}
+
+func (a cat) Aggr() interface{} {
+	v := make(map[string]uint32, len(a))
+	for b, c := range a {
+		v[b] = c
+		a[b] = 0
+	}
+	return v
+}

aggr/count.go

@@ -0,0 +1,16 @@
+package aggr
+
+type count struct {
+	c uint32
+}
+
+func (a *count) Push(v interface{}) error {
+	a.c++
+	return nil
+}
+
+func (a *count) Aggr() interface{} {
+	v := a.c
+	a.c = 0
+	return v
+}

aggr/field.go

@@ -0,0 +1,115 @@
+package aggr
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/elgs/gojq"
+)
+
+// Field describe an aggregation on a field.
+type Field struct {
+	Path  string
+	Name  string
+	Aggrs map[string]Aggregator
+}
+
+type Fields struct {
+	f  []Field
+	mu sync.Mutex
+}
+
+func NewFields(defs []string) (*Fields, error) {
+	fields := make([]Field, 0, len(defs))
+	for _, def := range defs {
+		f, err := NewField(def)
+		if err != nil {
+			return nil, fmt.Errorf("%s: %v", def, err)
+		}
+		fields = append(fields, f)
+	}
+	return &Fields{f: fields}, nil
+}
+
+func (fs *Fields) Push(jq *gojq.JQ) error {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	for _, f := range fs.f {
+		if err := f.Push(jq); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (fs *Fields) Aggr() map[string]interface{} {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	v := map[string]interface{}{}
+	for _, f := range fs.f {
+		v[f.Name] = f.Aggr()
+	}
+	return v
+}
+
+// NewField parses a field definition.
+func NewField(def string) (Field, error) {
+	if strings.HasPrefix(def, "@count") {
+		name := def
+		if idx := strings.LastIndexByte(name, '='); idx != -1 {
+			name = name[idx+1:]
+		}
+		return Field{
+			Path:  ".",
+			Name:  name,
+			Aggrs: map[string]Aggregator{"": &count{}},
+		}, nil
+	}
+	idx := strings.LastIndexByte(def, ':')
+	if idx == -1 {
+		return Field{}, errors.New("missing aggregation definition")
+	}
+	path := def[idx+1:]
+	p := &aggrsParser{exp: def[:idx]}
+	aggrs, err := p.parse()
+	if err != nil {
+		return Field{}, err
+	}
+	f := Field{
+		Path:  path,
+		Name:  path,
+		Aggrs: aggrs,
+	}
+	if idx = strings.LastIndexByte(f.Path, '='); idx != -1 {
+		f.Name = f.Path[idx+1:]
+		f.Path = f.Path[:idx]
+	}
+	return f, nil
+}
+
+func (f *Field) Push(jq *gojq.JQ) error {
+	v, err := jq.Query(f.Path)
+	if err != nil {
+		return err
+	}
+	for name, aggr := range f.Aggrs {
+		if err := aggr.Push(v); err != nil {
+			return fmt.Errorf("%s: %v", name, err)
+		}
+	}
+	return nil
+}
+
+func (f *Field) Aggr() interface{} {
+	if f.Path == "." && f.Aggrs[""] != nil {
+		// Count special field
+		return f.Aggrs[""].Aggr()
+	}
+	v := map[string]interface{}{}
+	for name, aggr := range f.Aggrs {
+		v[name] = aggr.Aggr()
+	}
+	return v
+}

aggr/hist.go

@@ -0,0 +1,61 @@
+package aggr
+
+import (
+	"sort"
+)
+
+type bucket struct {
+	name string
+	val  float64
+}
+
+type buckets []bucket
+
+func (b buckets) Len() int           { return len(b) }
+func (b buckets) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
+func (b buckets) Less(i, j int) bool { return b[i].val < b[j].val }
+
+type hist struct {
+	buckets buckets
+	counts  []uint32
+}
+
+func newHist(args []string) (Aggregator, error) {
+	h := hist{
+		buckets: make([]bucket, 0, len(args)),
+		counts:  make([]uint32, len(args)),
+	}
+	for _, arg := range args {
+		val, err := parseFloat(arg)
+		if err != nil {
+			return nil, err
+		}
+		h.buckets = append(h.buckets, bucket{name: arg, val: val})
+	}
+	sort.Sort(h.buckets)
+	return h, nil
+}
+
+func (a hist) Push(v interface{}) error {
+	f, err := parseFloat(v)
+	if err != nil {
+		return err
+	}
+	var i int
+	for ; i < len(a.buckets)-1; i++ {
+		if f >= a.buckets[i].val && f < a.buckets[i+1].val {
+			break
+		}
+	}
+	a.counts[i]++
+	return nil
+}
+
+func (a hist) Aggr() interface{} {
+	v := make(map[string]uint32, len(a.buckets))
+	for i, b := range a.buckets {
+		v[b.name] = a.counts[i]
+		a.counts[i] = 0
+	}
+	return v
+}