feat: add sum aggregator to support summing numeric values

README.md

@@ -31,7 +31,7 @@ mkr plugin install --upgrade monitoring-forge/mackerel-plugin-jsonl
 | `--version` | `-v` | バージョン情報を表示します。 |
 | `--key-name` | `-k` | メトリック名に使用するキー名を指定します（必須）。複数指定できます。 |
 | `--json-key` | `-j` | ログから値を抽出するJSONキーを指定します（必須）。パイプ区切りで変換関数を指定できます。複数指定できます。 |
-| `--aggregator` | `-a` | 集計方法を指定します（必須）。`count`、`group_by`、`group_by_with_percentage`、`percentile`、`percentile(...)` が指定可能です。複数指定できます。 |
+| `--aggregator` | `-a` | 集計方法を指定します（必須）。`count`、`sum`、`group_by`、`group_by_with_percentage`、`percentile`、`percentile(...)` が指定可能です。複数指定できます。 |
 | `--log-file` | `-l` | 読み込むJSON Lines形式のログファイルパスを指定します（必須）。 |
 | `--prefix` | | Mackerelメトリック名のプレフィックスを指定します（必須）。 |
 | `--filter` | | ログ行をフィルタする文字列を指定します。指定した文字列を含む行のみ対象となります。 |
@@ -55,6 +55,8 @@ command = "/path/to/mackerel-plugin-jsonl --key-name count --json-key foo.bar --
 
 - `count` : 対象となるキーが存在する行をカウントします
 	- 例: `json.total.count        80000.000000    1760281381`
+- `sum` : 指定した数値キーを合計します
+  - 例: `json.bytes_sent.sum        75000.000000    1760281381`
 - `group_by` : 指定したキーの値ごとに件数を集計します。
 	- 例: `json.status.2xx 24769.733333    1760281381`
 - `group_by_with_percentage` : group_byの各値ごとの割合（%）も出力します。
@@ -103,13 +105,12 @@ command = "/path/to/mackerel-plugin-jsonl --key-name count --json-key foo.bar --
 | `trimspace` | 値の前後の空白を除去します。 | `group_by`, `group_by_with_percentage` |
 | `replace('pattern','repl')` | 正規表現で値を置換します。 | `group_by`, `group_by_with_percentage` |
 | `have('key1','key2','key3')` | 集計結果に指定したキーをあらかじめ `0` で追加します。集計した値がない場合でもメトリックを出力できます。 | `group_by`, `group_by_with_percentage` |
-| `multiply(factor)` | 数値に指定した係数を掛けます。 | `count`, `group_by`, `group_by_with_percentage`, `percentile` |
+| `multiply(factor)` | 数値に指定した係数を掛けます。 | `count`, `sum`, `group_by`, `group_by_with_percentage`, `percentile` |
 
 ### 注意事項
 
-- 文字列変換関数（`tolower`、`toupper`、`trimspace`、`replace`）と数値変換関数（`multiply`）は、すべての aggregator で利用できます。
-- ただし、`have()` は `group_by` および `group_by_with_percentage` のみで利用できます。
-- `count` や `percentile` で `have()` は指定できません。
+- 数値変換関数（`multiply`）は、すべての aggregator で利用できます。
+- 文字列変換関数（`tolower`、`toupper`、`trimspace`、`replace`）と `have()` は `group_by` および `group_by_with_percentage` のみで利用できます。
 - `multiply()` は `count` では件数、`group_by`/`group_by_with_percentage` では各値の件数、`percentile` では各統計値に係数を掛けます。
 
 例:

cmd/mackerel-plugin-jsonl/main.go

@@ -17,7 +17,7 @@ type Opt struct {
 	Ignore              string   `long:"ignore" description:"ignore string used before check pattern."`
 	KeyNames            []string `short:"k" long:"key-name" required:"true" description:"Key name for json path"`
 	JsonKeys            []string `short:"j" long:"json-key" required:"true" description:"JSON key and modifier functions to extract log message."`
-	Aggregator          []string `short:"a" long:"aggregator" required:"true" description:"Aggregator type: count, group_by, group_by_with_percentage, percentile, or percentile(max,min,mean,50,99.9)."` // nolint:staticcheck
+	Aggregator          []string `short:"a" long:"aggregator" required:"true" description:"Aggregator type: count, sum, group_by, group_by_with_percentage, percentile, or percentile(max,min,mean,50,99.9)."` // nolint:staticcheck
 	SkipUntilBracket    bool     `long:"skip-until-json" description:"skip reading until first { for json log with plain text header"`
 	Prefix              string   `long:"prefix" required:"true" description:"Metric key prefix"`
 	PerSec              bool     `long:"per-second" description:"calculate per-seconds count. default per minute count"`

cmd/mackerel-plugin-jsonl/parser_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
@@ -118,6 +119,11 @@ func TestParser_Parse(t *testing.T) {
 				jsonKey:    []string{"foo"},
 				count:      0,
 			},
+			{
+				aggregator: "sum",
+				jsonKey:    []string{"ptime"},
+				sum:        0,
+			},
 			{
 				aggregator: "group_by",
 				jsonKey:    []string{"status"},
@@ -129,33 +135,41 @@ func TestParser_Parse(t *testing.T) {
 				percentiles: sampdo.New(sampdo.WithInitialCapacity(1024)),
 			},
 		},
-		paths: [][]string{{"foo"}, {"status"}, {"ptime"}},
+		paths: [][]string{{"foo"}, {"ptime"}, {"status"}, {"ptime"}},
 	}
 
-	json := []byte(`{"foo": 1, "status": "ok", "ptime": 100}`)
-	err := opt.Parse(json)
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
+	statuses := []string{"ok", "ng", "unknown"}
+	for i := range 100 {
+		status := statuses[i%len(statuses)]
+
+		jsonStr := fmt.Sprintf(`{"foo": 1, "status": "%s", "ptime": %d}`, status, 100+i)
+		err := opt.Parse([]byte(jsonStr))
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+	}
+	if opt.aggregatorFunctions[0].count != 100 {
+		t.Errorf("expected count 100, got %v", opt.aggregatorFunctions[0].count)
 	}
-	if opt.aggregatorFunctions[0].count != 1 {
-		t.Errorf("expected count 1, got %v", opt.aggregatorFunctions[0].count)
+	if opt.aggregatorFunctions[1].sum != 14950 {
+		t.Errorf("expected sum 14950, got %v", opt.aggregatorFunctions[1].sum)
 	}
-	if opt.aggregatorFunctions[1].groupBy["ok"] != 1 {
-		t.Errorf("expected groupBy ok 1, got %v", opt.aggregatorFunctions[1].groupBy["ok"])
+	if opt.aggregatorFunctions[2].groupBy["ok"] != 34 {
+		t.Errorf("expected groupBy ok 34, got %v", opt.aggregatorFunctions[2].groupBy["ok"])
 	}
-	sorted, err := opt.aggregatorFunctions[2].percentiles.Sorted()
+	sorted, err := opt.aggregatorFunctions[3].percentiles.Sorted()
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	if sorted.Count() != 1 {
-		t.Errorf("expected percentiles count 1, got %v", sorted.Count())
+	if sorted.Count() != 100 {
+		t.Errorf("expected percentiles count 100, got %v", sorted.Count())
 	}
 	maxPtime, err := sorted.Max()
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	if maxPtime != 100 {
-		t.Errorf("expected percentiles max 100, got %v", maxPtime)
+	if maxPtime != 199 {
+		t.Errorf("expected percentiles max 199, got %v", maxPtime)
 	}
 }
 

cmd/mackerel-plugin-jsonl/percentile.go

@@ -17,7 +17,7 @@ type percentileTarget struct {
 
 func parseAggregator(s string) (string, []percentileTarget, error) {
 	switch s {
-	case "count", "group_by", "group_by_with_percentage":
+	case "count", "sum", "group_by", "group_by_with_percentage":
 		return s, nil, nil
 	case "percentile":
 		return s, []percentileTarget{{name: "mean"}, {name: "p90", value: 90}, {name: "p95", value: 95}, {name: "p99", value: 99}}, nil

cmd/mackerel-plugin-jsonl/reader.go

@@ -20,6 +20,7 @@ type AggregatorFunction struct {
 	JsonKeyInitializers     []JsonKeyInitializer
 	aggregator              string
 	count                   int
+	sum                     float64
 	groupBy                 map[string]int
 	percentiles             *sampdo.Sampdo
 	percentileTargets       []percentileTarget
@@ -52,11 +53,15 @@ func (af *AggregatorFunction) appendData(b []byte) error {
 		af.count++
 	case "group_by", "group_by_with_percentage":
 		af.groupBy[string(b)]++
-	case "percentile":
+	case "percentile", "sum":
 		floatValue, err := ltsvparser.ParseFloat(b)
 		if err != nil {
 			return err
 		}
+		if af.aggregator == "sum" {
+			af.sum += floatValue
+			return nil
+		}
 		err = af.percentiles.Append(floatValue)
 		if err != nil {
 			return err
@@ -105,7 +110,7 @@ func (p *Opt) buildAggregatorFunction(i int) (*AggregatorFunction, error) {
 		return nil, err
 	}
 	switch aggregator {
-	case "count", "percentile":
+	case "count", "sum", "percentile":
 		if len(modifiers) > 0 || len(initializers) > 0 {
 			return nil, fmt.Errorf("modifiers and initializers are not supported for %s aggregator", p.Aggregator[i])
 		}
@@ -177,6 +182,8 @@ func (p *Opt) writeAggregatorOutput(output *strings.Builder, af *AggregatorFunct
 	switch af.aggregator {
 	case "count":
 		p.writeCountOutput(output, af, now)
+	case "sum":
+		p.writeSumOutput(output, af, now)
 	case "group_by", "group_by_with_percentage":
 		p.writeGroupByOutput(output, af, now)
 	case "percentile":
@@ -193,6 +200,15 @@ func (p *Opt) writeCountOutput(output *strings.Builder, af *AggregatorFunction,
 	fmt.Fprintf(output, "%s.%s\t%f\t%d\n", p.Prefix, af.name, p.calculatePerDuration(count), now)
 }
 
+func (p *Opt) writeSumOutput(output *strings.Builder, af *AggregatorFunction, now uint64) {
+	if p.duration == 0 {
+		// avoid division by zero
+		return
+	}
+	sum := af.applyNumericModifiers(af.sum)
+	fmt.Fprintf(output, "%s.%s\t%f\t%d\n", p.Prefix, af.name, p.calculatePerDuration(sum), now)
+}
+
 func (p *Opt) writeGroupByOutput(output *strings.Builder, af *AggregatorFunction, now uint64) {
 	if p.duration == 0 {
 		// avoid division by zero