New Upstream Snapshot - golang-collectd
Ready changes
Summary
Merged new upstream version: 0.5.0+git20200608.1.92e86f9 (was: 0.3.0+git20181025.f80706d).
Resulting package
Built on 2022-04-16T22:47 (took 5m5s)
The resulting binary packages can be installed (if you have the apt repository enabled) by running one of:
apt install -t fresh-snapshots golang-collectd-dev
Lintian Result
Diff
diff --git a/.travis.yml b/.travis.yml
index 7cc2842..438cf32 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,13 +1,17 @@
+# Request newer Ubuntu version. Xenial (current default) ships collectd 5.5, which is too old.
+dist: bionic
language: go
go:
- - "1.11"
- - "1.10"
+ - "stable"
+ # - "oldstable" # https://github.com/travis-ci/gimme/issues/179
- "master"
+before_install:
+ - sudo apt-get -y install collectd-dev
+
env:
- # The "plugin" package required the collectd sources, which are not (yet) available.
- - CGO_ENABLED=0
+ - CGO_ENABLED=1 CGO_CPPFLAGS="-I/usr/include/collectd/core/daemon -I/usr/include/collectd/core -I/usr/include/collectd"
go_import_path: collectd.org
@@ -19,8 +23,10 @@ before_script:
- go get golang.org/x/lint/golint
script:
- - go vet -x ./...
- - go test -v ./...
+ - go test -coverprofile=/dev/null ./...
+ - ./check_fmt.sh
+ - go vet ./...
+ - golint -set_exit_status {cdtime,config,exec,export,format,meta,network}/... plugin/ rpc/
# TODO(octo): run the fuzz test:
# - go test -v -tags gofuzz ./...
diff --git a/api/json.go b/api/json.go
index 87ba36e..3884933 100644
--- a/api/json.go
+++ b/api/json.go
@@ -5,6 +5,7 @@ import (
"fmt"
"collectd.org/cdtime"
+ "collectd.org/meta"
)
// jsonValueList represents the format used by collectd's JSON export.
@@ -19,6 +20,7 @@ type jsonValueList struct {
PluginInstance string `json:"plugin_instance,omitempty"`
Type string `json:"type"`
TypeInstance string `json:"type_instance,omitempty"`
+ Meta meta.Data `json:"meta,omitempty"`
}
// MarshalJSON implements the "encoding/json".Marshaler interface for
@@ -35,6 +37,7 @@ func (vl *ValueList) MarshalJSON() ([]byte, error) {
PluginInstance: vl.PluginInstance,
Type: vl.Type,
TypeInstance: vl.TypeInstance,
+ Meta: vl.Meta,
}
for i, v := range vl.Values {
@@ -115,5 +118,7 @@ func (vl *ValueList) UnmarshalJSON(data []byte) error {
copy(vl.DSNames, jvl.DSNames)
}
+ vl.Meta = jvl.Meta
+
return nil
}
diff --git a/api/json_test.go b/api/json_test.go
index 5786c9c..8b45327 100644
--- a/api/json_test.go
+++ b/api/json_test.go
@@ -8,6 +8,9 @@ import (
"reflect"
"testing"
"time"
+
+ "collectd.org/meta"
+ "github.com/google/go-cmp/cmp"
)
func TestValueList(t *testing.T) {
@@ -21,13 +24,19 @@ func TestValueList(t *testing.T) {
Interval: 10 * time.Second,
Values: []Value{Gauge(42)},
DSNames: []string{"legacy"},
+ Meta: meta.Data{
+ "foo": meta.String("bar"),
+ },
}
- want := `{"values":[42],"dstypes":["gauge"],"dsnames":["legacy"],"time":1426585562.999,"interval":10.000,"host":"example.com","plugin":"golang","type":"gauge"}`
+ want := `{"values":[42],"dstypes":["gauge"],"dsnames":["legacy"],"time":1426585562.999,"interval":10.000,"host":"example.com","plugin":"golang","type":"gauge","meta":{"foo":"bar"}}`
got, err := vlWant.MarshalJSON()
- if err != nil || string(got) != want {
- t.Errorf("got (%s, %v), want (%s, nil)", got, err, want)
+ if err != nil {
+ t.Fatalf("ValueList.MarshalJSON() = %v", err)
+ }
+ if diff := cmp.Diff(want, string(got)); diff != "" {
+ t.Errorf("ValueList.MarshalJSON() differs (+got/-want):\n%s", diff)
}
var vlGot ValueList
diff --git a/api/main.go b/api/main.go
index 96496b1..ea047a0 100644
--- a/api/main.go
+++ b/api/main.go
@@ -4,10 +4,13 @@ package api // import "collectd.org/api"
import (
"context"
"fmt"
- "log"
"strconv"
"strings"
+ "sync"
"time"
+
+ "collectd.org/meta"
+ "go.uber.org/multierr"
)
// Value represents either a Gauge or a Derive. It is Go's equivalent to the C
@@ -85,6 +88,7 @@ type ValueList struct {
Interval time.Duration
Values []Value
DSNames []string
+ Meta meta.Data
}
// DSName returns the name of the data source at the given index. If vl.DSNames
@@ -100,12 +104,42 @@ func (vl *ValueList) DSName(index int) string {
return "value"
}
+// Clone returns a copy of vl.
+// Unfortunately, many functions expect a pointer to a value list. If the
+// original value list must not be modified, it may be necessary to create and
+// pass a copy. This is what this method helps to do.
+func (vl *ValueList) Clone() *ValueList {
+ if vl == nil {
+ return nil
+ }
+
+ vlCopy := *vl
+
+ vlCopy.Values = make([]Value, len(vl.Values))
+ copy(vlCopy.Values, vl.Values)
+
+ vlCopy.DSNames = make([]string, len(vl.DSNames))
+ copy(vlCopy.DSNames, vl.DSNames)
+
+ vlCopy.Meta = vl.Meta.Clone()
+
+ return &vlCopy
+}
+
// Writer are objects accepting a ValueList for writing, for example to the
// network.
type Writer interface {
Write(context.Context, *ValueList) error
}
+// WriterFunc implements the Writer interface based on a wrapped function.
+type WriterFunc func(context.Context, *ValueList) error
+
+// Write calls the wrapped function.
+func (f WriterFunc) Write(ctx context.Context, vl *ValueList) error {
+ return f(ctx, vl)
+}
+
// String returns a string representation of the Identifier.
func (id Identifier) String() string {
str := id.Host + "/" + id.Plugin
@@ -119,36 +153,56 @@ func (id Identifier) String() string {
return str
}
-// Dispatcher implements a multiplexer for Writer, i.e. each ValueList
-// written to it is copied and written to each registered Writer.
-type Dispatcher struct {
- writers []Writer
-}
-
-// Add adds a Writer to the Dispatcher.
-func (d *Dispatcher) Add(w Writer) {
- d.writers = append(d.writers, w)
-}
-
-// Len returns the number of Writers belonging to the Dispatcher.
-func (d *Dispatcher) Len() int {
- return len(d.writers)
-}
-
-// Write starts a new Goroutine for each Writer which creates a copy of the
-// ValueList and then calls the Writer with the copy. It returns nil
-// immediately.
-func (d *Dispatcher) Write(ctx context.Context, vl *ValueList) error {
- for _, w := range d.writers {
+// Fanout implements a multiplexer for Writer, i.e. each ValueList written to
+// it is copied and written to each Writer.
+type Fanout []Writer
+
+// Write writes the value list to each writer. Each writer receives a copy of
+// the value list to avoid writers interfering with one another. Writers are
+// executed concurrently. Write blocks until all writers have returned and
+// returns an error containing all errors returned by writers.
+//
+// If the context is canceled, Write returns an error immediately. Since it may
+// return before all writers have finished, the returned error may not contain
+// the error of all writers.
+func (f Fanout) Write(ctx context.Context, vl *ValueList) error {
+ var (
+ ch = make(chan error)
+ wg sync.WaitGroup
+ )
+
+ for _, w := range f {
+ wg.Add(1)
go func(w Writer) {
- vlCopy := vl
- vlCopy.Values = make([]Value, len(vl.Values))
- copy(vlCopy.Values, vl.Values)
-
- if err := w.Write(ctx, vlCopy); err != nil {
- log.Printf("%T.Write(): %v", w, err)
+ defer wg.Done()
+
+ if err := w.Write(ctx, vl.Clone()); err != nil {
+ // block until the error is read, or until the
+ // context is canceled.
+ select {
+ case ch <- fmt.Errorf("%T.Write(): %w", w, err):
+ case <-ctx.Done():
+ }
}
}(w)
}
- return nil
+
+ go func() {
+ wg.Wait()
+ close(ch)
+ }()
+
+ var errs error
+ for {
+ select {
+ case err, ok := <-ch:
+ if !ok {
+ // channel closed, all goroutines done
+ return errs
+ }
+ errs = multierr.Append(errs, err)
+ case <-ctx.Done():
+ return multierr.Append(errs, ctx.Err())
+ }
+ }
}
diff --git a/api/main_test.go b/api/main_test.go
index a2de72d..7b86ba8 100644
--- a/api/main_test.go
+++ b/api/main_test.go
@@ -1,17 +1,23 @@
-package api // import "collectd.org/api"
+package api_test
import (
+ "context"
+ "errors"
+ "sync"
"testing"
+
+ "collectd.org/api"
+ "github.com/google/go-cmp/cmp"
)
func TestParseIdentifier(t *testing.T) {
cases := []struct {
Input string
- Want Identifier
+ Want api.Identifier
}{
{
Input: "example.com/golang/gauge",
- Want: Identifier{
+ Want: api.Identifier{
Host: "example.com",
Plugin: "golang",
Type: "gauge",
@@ -19,7 +25,7 @@ func TestParseIdentifier(t *testing.T) {
},
{
Input: "example.com/golang-foo/gauge-bar",
- Want: Identifier{
+ Want: api.Identifier{
Host: "example.com",
Plugin: "golang",
PluginInstance: "foo",
@@ -29,7 +35,7 @@ func TestParseIdentifier(t *testing.T) {
},
{
Input: "example.com/golang-a-b/gauge-b-c",
- Want: Identifier{
+ Want: api.Identifier{
Host: "example.com",
Plugin: "golang",
PluginInstance: "a-b",
@@ -40,7 +46,7 @@ func TestParseIdentifier(t *testing.T) {
}
for i, c := range cases {
- if got, err := ParseIdentifier(c.Input); got != c.Want || err != nil {
+ if got, err := api.ParseIdentifier(c.Input); got != c.Want || err != nil {
t.Errorf("case %d: got (%v, %v), want (%v, %v)", i, got, err, c.Want, nil)
}
}
@@ -51,14 +57,14 @@ func TestParseIdentifier(t *testing.T) {
}
for _, c := range failures {
- if got, err := ParseIdentifier(c); err == nil {
- t.Errorf("got (%v, %v), want (%v, !%v)", got, err, Identifier{}, nil)
+ if got, err := api.ParseIdentifier(c); err == nil {
+ t.Errorf("got (%v, %v), want (%v, !%v)", got, err, api.Identifier{}, nil)
}
}
}
func TestIdentifierString(t *testing.T) {
- id := Identifier{
+ id := api.Identifier{
Host: "example.com",
Plugin: "golang",
Type: "gauge",
@@ -84,3 +90,123 @@ func TestIdentifierString(t *testing.T) {
}
}
}
+
+type testWriter struct {
+ got *api.ValueList
+ wg *sync.WaitGroup
+ ch chan struct{}
+ err error
+}
+
+func (w *testWriter) Write(ctx context.Context, vl *api.ValueList) error {
+ w.got = vl
+ w.wg.Done()
+
+ select {
+ case <-w.ch:
+ return w.err
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
+type testError struct{}
+
+func (testError) Error() string {
+ return "test error"
+}
+
+func TestFanout(t *testing.T) {
+ cases := []struct {
+ title string
+ returnError bool
+ cancelContext bool
+ }{
+ {
+ title: "success",
+ },
+ {
+ title: "error",
+ returnError: true,
+ },
+ {
+ title: "context canceled",
+ cancelContext: true,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ var (
+ done = make(chan struct{})
+ wg sync.WaitGroup
+ )
+
+ var writerError error
+ if tc.returnError {
+ writerError = testError{}
+ }
+ writers := []*testWriter{
+ {
+ wg: &wg,
+ ch: done,
+ err: writerError,
+ },
+ {
+ wg: &wg,
+ ch: done,
+ err: writerError,
+ },
+ }
+ wg.Add(len(writers))
+
+ go func() {
+ // wait for all writers to be called, then signal them to return
+ wg.Wait()
+
+ if tc.cancelContext {
+ cancel()
+ } else {
+ close(done)
+ }
+ }()
+
+ want := &api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestFanout",
+ Type: "gauge",
+ },
+ Values: []api.Value{api.Gauge(42)},
+ DSNames: []string{"value"},
+ }
+
+ var f api.Fanout
+ for _, w := range writers {
+ f = append(f, w)
+ }
+
+ err := f.Write(ctx, want)
+ switch {
+ case tc.returnError && !errors.Is(err, testError{}):
+ t.Errorf("Fanout.Write() = %v, want %T", err, testError{})
+ case tc.cancelContext && !errors.Is(err, context.Canceled):
+ t.Errorf("Fanout.Write() = %v, want %T", err, context.Canceled)
+ case !tc.returnError && !tc.cancelContext && err != nil:
+ t.Errorf("Fanout.Write() = %v", err)
+ }
+
+ for i, w := range writers {
+ if want == w.got {
+ t.Errorf("writers[%d].vl == w.got, want copy", i)
+ }
+ if diff := cmp.Diff(want, w.got); diff != "" {
+ t.Errorf("writers[%d].vl differs (+got/-want):\n%s", i, diff)
+ }
+ }
+ })
+ }
+}
diff --git a/api/types_test.go b/api/types_test.go
index 71234da..fd932bf 100644
--- a/api/types_test.go
+++ b/api/types_test.go
@@ -25,7 +25,7 @@ mysql_qcache hits:COUNTER:0:U, inserts:COUNTER:0:U, not_cached:COUNTER:0:U, low
want := &DataSet{
Name: "percent",
Sources: []DataSource{
- DataSource{
+ {
Name: "value",
Type: reflect.TypeOf(Gauge(0)),
Min: 0.0,
diff --git a/cdtime/cdtime.go b/cdtime/cdtime.go
index 3fd8060..f373621 100644
--- a/cdtime/cdtime.go
+++ b/cdtime/cdtime.go
@@ -14,6 +14,9 @@ type Time uint64
// New returns a new Time representing time t.
func New(t time.Time) Time {
+ if t.IsZero() {
+ return 0
+ }
return newNano(uint64(t.UnixNano()))
}
@@ -24,6 +27,10 @@ func NewDuration(d time.Duration) Time {
// Time converts and returns the time as time.Time.
func (t Time) Time() time.Time {
+ if t == 0 {
+ return time.Time{}
+ }
+
s, ns := t.decompose()
return time.Unix(s, ns)
}
diff --git a/cdtime/cdtime_test.go b/cdtime/cdtime_test.go
index 3fbfcba..fcc12cd 100644
--- a/cdtime/cdtime_test.go
+++ b/cdtime/cdtime_test.go
@@ -1,13 +1,16 @@
-package cdtime // import "collectd.org/cdtime"
+package cdtime_test
import (
+ "encoding/json"
"testing"
"time"
+
+ "collectd.org/cdtime"
)
-// TestConversion converts a time.Time to a cdtime.Time and back, expecting the
+// TestNew converts a time.Time to a cdtime.Time and back, expecting the
// original time.Time back.
-func TestConversion(t *testing.T) {
+func TestNew(t *testing.T) {
cases := []string{
"2009-02-04T21:00:57-08:00",
"2009-02-04T21:00:57.1-08:00",
@@ -28,54 +31,73 @@ func TestConversion(t *testing.T) {
continue
}
- cdtime := New(want)
- got := cdtime.Time()
+ ct := cdtime.New(want)
+ got := ct.Time()
if !got.Equal(want) {
t.Errorf("cdtime.Time(): got %v, want %v", got, want)
}
}
}
-func TestDecompose(t *testing.T) {
- cases := []struct {
- in Time
- s, ns int64
- }{
- // 1546167635576736987 / 2^30 = 1439980823.1524536265...
- {Time(1546167635576736987), 1439980823, 152453627},
- // 1546167831554815222 / 2^30 = 1439981005.6712620165...
- {Time(1546167831554815222), 1439981005, 671262017},
- // 1546167986577716567 / 2^30 = 1439981150.0475896215...
- {Time(1546167986577716567), 1439981150, 47589622},
+func TestNew_zero(t *testing.T) {
+ var (
+ got = cdtime.New(time.Time{})
+ want = cdtime.Time(0)
+ )
+ if got != want {
+ t.Errorf("cdtime.New(time.Time{}) = %v, want %v", got, want)
}
- for _, c := range cases {
- s, ns := c.in.decompose()
+ if got := cdtime.Time(0).Time(); !got.IsZero() {
+ t.Errorf("cdtime.Time(0).Time() = %v, want zero value (%v)", got, time.Time{})
+ }
+}
- if s != c.s || ns != c.ns {
- t.Errorf("decompose(%d) = (%d, %d) want (%d, %d)", c.in, s, ns, c.s, c.ns)
- }
+func TestMarshalJSON(t *testing.T) {
+ tm := time.Unix(1587671455, 499000000)
+
+ orig := cdtime.New(tm)
+ data, err := json.Marshal(orig)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ var got cdtime.Time
+ if err := json.Unmarshal(data, &got); err != nil {
+ t.Fatal(err)
+ }
+
+ // JSON Marshaling is not loss-less, because it only encodes
+ // millisecond precision.
+ if got, want := got.String(), "1587671455.499"; got != want {
+ t.Errorf("json.Unmarshal() result differs: got %q, want %q", got, want)
}
}
-func TestNewNano(t *testing.T) {
+func TestNewDuration(t *testing.T) {
cases := []struct {
- ns uint64
- want Time
+ d time.Duration
+ want cdtime.Time
}{
// 1439981652801860766 * 2^30 / 10^9 = 1546168526406004689.4
- {1439981652801860766, Time(1546168526406004689)},
+ {1439981652801860766 * time.Nanosecond, cdtime.Time(1546168526406004689)},
// 1439981836985281914 * 2^30 / 10^9 = 1546168724171447263.4
- {1439981836985281914, Time(1546168724171447263)},
+ {1439981836985281914 * time.Nanosecond, cdtime.Time(1546168724171447263)},
// 1439981880053705608 * 2^30 / 10^9 = 1546168770415815077.4
- {1439981880053705608, Time(1546168770415815077)},
+ {1439981880053705608 * time.Nanosecond, cdtime.Time(1546168770415815077)},
+ // 1439981880053705920 * 2^30 / 10^9 = 1546168770415815412.5
+ {1439981880053705920 * time.Nanosecond, cdtime.Time(1546168770415815413)},
+ {0, 0},
}
- for _, c := range cases {
- got := newNano(c.ns)
+ for _, tc := range cases {
+ d := cdtime.NewDuration(tc.d)
+ if got, want := d, tc.want; got != want {
+ t.Errorf("NewDuration(%v) = %d, want %d", tc.d, got, want)
+ }
- if got != c.want {
- t.Errorf("newNano(%d) = %d, want %d", c.ns, got, c.want)
+ if got, want := d.Duration(), tc.d; got != want {
+ t.Errorf("%#v.Duration() = %v, want %v", d, got, want)
}
}
}
diff --git a/check_fmt.sh b/check_fmt.sh
new file mode 100755
index 0000000..9d445e8
--- /dev/null
+++ b/check_fmt.sh
@@ -0,0 +1,11 @@
+#!/bin/bash
+
+declare -r NEED_FMT="$(gofmt -l **/*.go)"
+
+if [[ -z "${NEED_FMT}" ]]; then
+ exit 0
+fi
+
+echo "The following files are NOT formatted with gofmt:"
+echo "${NEED_FMT}"
+exit 1
diff --git a/config/config.go b/config/config.go
new file mode 100644
index 0000000..bcba91b
--- /dev/null
+++ b/config/config.go
@@ -0,0 +1,408 @@
+/*
+Package config provides types that represent a plugin's configuration.
+
+The types provided in this package are fairly low level and correspond directly
+to types in collectd:
+
+· "Block" corresponds to "oconfig_item_t".
+
+· "Value" corresponds to "oconfig_value_t".
+
+Blocks contain a Key, and optionally Values and/or children (nested Blocks). In
+collectd's configuration, these pieces are represented as follows:
+
+ <Key "Value">
+ Child "child value"
+ </Key>
+
+In Go, this would be represented as:
+
+ Block{
+ Key: "Key",
+ Values: []Value{String("Value")},
+ Children: []Block{
+ {
+ Key: "Child",
+ Values: []Value{String("child value")},
+ },
+ },
+ }
+
+The recommended way to work with configurations is to define a data type
+representing the configuration, then use "Block.Unmarshal" to map the Block
+representation onto the data type.
+*/
+package config // import "collectd.org/config"
+
+import (
+ "bytes"
+ "fmt"
+ "math"
+ "net"
+ "reflect"
+ "strings"
+
+ "github.com/google/go-cmp/cmp"
+)
+
+type valueType int
+
+const (
+ stringType valueType = iota
+ numberType
+ booleanType
+)
+
+// Value may be either a string, float64 or boolean value.
+// This is the Go equivalent of the C type "oconfig_value_t".
+type Value struct {
+ typ valueType
+ s string
+ f float64
+ b bool
+}
+
+// String returns a new string Value.
+func String(v string) Value { return Value{typ: stringType, s: v} }
+
+// Float64 returns a new float64 Value.
+func Float64(v float64) Value { return Value{typ: numberType, f: v} }
+
+// Bool returns a new bool Value.
+func Bool(v bool) Value { return Value{typ: booleanType, b: v} }
+
+// Values allocates and initializes a []Value slice. "string", "float64", and
+// "bool" are mapped directly. "[]byte" is converted to a string. Numeric types
+// (except complex numbers) are converted to float64. All other values are
+// converted to string using the `%v` format.
+func Values(values ...interface{}) []Value {
+ var ret []Value
+ for _, v := range values {
+ if v == nil {
+ ret = append(ret, Float64(math.NaN()))
+ continue
+ }
+
+ // check for exact matches first.
+ switch v := v.(type) {
+ case string:
+ ret = append(ret, String(v))
+ continue
+ case []byte:
+ ret = append(ret, String(string(v)))
+ continue
+ case bool:
+ ret = append(ret, Bool(v))
+ continue
+ }
+
+ // Handle numerical types that can be converted to float64:
+ var (
+ valueType = reflect.TypeOf(v)
+ float64Type = reflect.TypeOf(float64(0))
+ )
+ if valueType.ConvertibleTo(float64Type) {
+ v := reflect.ValueOf(v).Convert(float64Type).Interface().(float64)
+ ret = append(ret, Float64(v))
+ continue
+ }
+
+ // Last resort: convert to a string using the "fmt" package:
+ ret = append(ret, String(fmt.Sprintf("%v", v)))
+ }
+ return ret
+}
+
+// GoString returns a Go statement for creating cv.
+func (cv Value) GoString() string {
+ switch cv.typ {
+ case stringType:
+ return fmt.Sprintf("config.String(%q)", cv.s)
+ case numberType:
+ return fmt.Sprintf("config.Float64(%v)", cv.f)
+ case booleanType:
+ return fmt.Sprintf("config.Bool(%v)", cv.b)
+ }
+ return "<invalid config.Value>"
+}
+
+// IsString returns true if cv is a string Value.
+func (cv Value) IsString() bool {
+ return cv.typ == stringType
+}
+
+// String returns Value as a string.
+// Non-string values are formatted according to their default format.
+func (cv Value) String() string {
+ return fmt.Sprintf("%v", cv.Interface())
+}
+
+// Float64 returns the value of a float64 Value.
+func (cv Value) Float64() (float64, bool) {
+ return cv.f, cv.typ == numberType
+}
+
+// Bool returns the value of a bool Value.
+func (cv Value) Bool() (bool, bool) {
+ return cv.b, cv.typ == booleanType
+}
+
+// Interface returns the specific value of Value without specifying its type,
+// useful for functions like fmt.Printf which can use variables with unknown
+// types.
+func (cv Value) Interface() interface{} {
+ switch cv.typ {
+ case stringType:
+ return cv.s
+ case numberType:
+ return cv.f
+ case booleanType:
+ return cv.b
+ }
+ return nil
+}
+
+func (cv Value) unmarshal(v reflect.Value) error {
+ rvt := v.Type()
+ var cvt reflect.Type
+ var cvv reflect.Value
+
+ switch cv.typ {
+ case stringType:
+ cvt = reflect.TypeOf(cv.s)
+ cvv = reflect.ValueOf(cv.s)
+ case booleanType:
+ cvt = reflect.TypeOf(cv.b)
+ cvv = reflect.ValueOf(cv.b)
+ case numberType:
+ cvt = reflect.TypeOf(cv.f)
+ cvv = reflect.ValueOf(cv.f)
+ default:
+ return fmt.Errorf("unexpected Value type: %v", cv.typ)
+ }
+
+ if cvt.ConvertibleTo(rvt) {
+ v.Set(cvv.Convert(rvt))
+ return nil
+ }
+ if v.Kind() == reflect.Slice && cvt.ConvertibleTo(rvt.Elem()) {
+ v.Set(reflect.Append(v, cvv.Convert(rvt.Elem())))
+ return nil
+ }
+ return fmt.Errorf("cannot unmarshal a %T to a %s", cv.Interface(), v.Type())
+}
+
+// Block represents one configuration block, which may contain other configuration blocks.
+type Block struct {
+ Key string
+ Values []Value
+ Children []Block
+}
+
+// IsZero returns true if b is the Zero value.
+func (b Block) IsZero() bool {
+ return b.Key == "" && len(b.Values) == 0 && len(b.Children) == 0
+}
+
+// Merge appends other's Children to b's Children. If Key or Values differ, an
+// error is returned.
+func (b *Block) Merge(other Block) error {
+ if b.IsZero() {
+ *b = other
+ return nil
+ }
+
+ if b.Key != other.Key || !cmp.Equal(b.Values, other.Values, cmp.AllowUnexported(Value{})) {
+ return fmt.Errorf("blocks differ: got {key:%v values:%v}, want {key:%v, values:%v}",
+ other.Key, other.Values, b.Key, b.Values)
+ }
+
+ b.Children = append(b.Children, other.Children...)
+ return nil
+}
+
+// Unmarshal applies the configuration from a Block to an arbitrary struct.
+func (b *Block) Unmarshal(v interface{}) error {
+ // If the target supports unmarshalling let it
+ if u, ok := v.(Unmarshaler); ok {
+ return u.UnmarshalConfig(*b)
+ }
+
+ // Sanity check value of the interface
+ rv := reflect.ValueOf(v)
+ if rv.Kind() != reflect.Ptr || rv.IsNil() {
+ return fmt.Errorf("can only unmarshal to a non-nil pointer") // TODO: better error message or nil if preferred
+ }
+
+ drv := rv.Elem() // get dereferenced value
+ drvk := drv.Kind()
+
+ // If config block has child blocks we can only unmarshal to a struct or slice of structs
+ if len(b.Children) > 0 {
+ if drvk != reflect.Struct && (drvk != reflect.Slice || drv.Type().Elem().Kind() != reflect.Struct) {
+ return fmt.Errorf("cannot unmarshal a config with children except to a struct or slice of structs")
+ }
+ }
+
+ switch drvk {
+ case reflect.Struct:
+ // Unmarshal values from config
+ if err := storeStructConfigValues(b.Values, drv); err != nil {
+ return fmt.Errorf("while unmarshalling config block values into %s: %s", drv.Type(), err)
+ }
+ for _, child := range b.Children {
+ // If a config has children but the struct has no corresponding field, or the corresponding field is an
+ // unexported struct field we throw an error.
+ if field := drv.FieldByName(child.Key); field.IsValid() && field.CanInterface() {
+ if err := child.Unmarshal(field.Addr().Interface()); err != nil {
+ // if err := child.Unmarshal(field.Interface()); err != nil {
+ return fmt.Errorf("in child config block %s: %s", child.Key, err)
+ }
+ } else {
+ return fmt.Errorf("found child config block with no corresponding field: %s", child.Key)
+ }
+ }
+ return nil
+ case reflect.Slice:
+ switch drv.Type().Elem().Kind() {
+ case reflect.Struct:
+ // Create a temporary Value of the same type as dereferenced value, then get a Value of the same type as
+ // its elements. Unmarshal into that Value and append the temporary Value to the original.
+ tv := reflect.New(drv.Type().Elem()).Elem()
+ if err := b.Unmarshal(tv.Addr().Interface()); err != nil {
+ return fmt.Errorf("unmarshaling into temporary value failed: %s", err)
+ }
+ drv.Set(reflect.Append(drv, tv))
+ return nil
+ default:
+ for _, cv := range b.Values {
+ tv := reflect.New(drv.Type().Elem()).Elem()
+ if err := cv.unmarshal(tv); err != nil {
+ return fmt.Errorf("while unmarhalling values into %s: %s", drv.Type(), err)
+ }
+ drv.Set(reflect.Append(drv, tv))
+ }
+ return nil
+ }
+ case reflect.String, reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Float32, reflect.Float64:
+ if len(b.Values) != 1 {
+ return fmt.Errorf("cannot unmarshal config option with %d values into scalar type %s", len(b.Values), drv.Type())
+ }
+ return b.Values[0].unmarshal(drv)
+ default:
+ return fmt.Errorf("cannot unmarshal into type %s", drv.Type())
+ }
+}
+
+func storeStructConfigValues(cvs []Value, v reflect.Value) error {
+ if len(cvs) == 0 {
+ return nil
+ }
+ args := v.FieldByName("Args")
+ if !args.IsValid() {
+ return fmt.Errorf("cannot unmarshal values to a struct without an Args field")
+ }
+ if len(cvs) > 1 && args.Kind() != reflect.Slice {
+ return fmt.Errorf("cannot unmarshal config block with multiple values to a struct with non-slice Args field")
+ }
+ for _, cv := range cvs {
+ if err := cv.unmarshal(args); err != nil {
+ return fmt.Errorf("while attempting to unmarshal config value \"%v\" in Args: %s", cv.Interface(), err)
+ }
+ }
+ return nil
+}
+
+// Unmarshaler is the interface implemented by types that can unmarshal a Block
+// representation of themselves.
+type Unmarshaler interface {
+ UnmarshalConfig(Block) error
+}
+
+// MarshalText produces a text version of Block. The result is parseable by collectd.
+// Implements the "encoding".TextMarshaler interface.
+func (b *Block) MarshalText() ([]byte, error) {
+ return b.marshalText("")
+}
+
+func (b *Block) marshalText(prefix string) ([]byte, error) {
+ var buf bytes.Buffer
+
+ values, err := valuesMarshalText(b.Values)
+ if err != nil {
+ return nil, err
+ }
+
+ if len(b.Children) == 0 {
+ fmt.Fprintf(&buf, "%s%s%s\n", prefix, b.Key, values)
+ return buf.Bytes(), nil
+ }
+
+ fmt.Fprintf(&buf, "%s<%s%s>\n", prefix, b.Key, values)
+ for _, c := range b.Children {
+ text, err := c.marshalText(prefix + " ")
+ if err != nil {
+ return nil, err
+ }
+ buf.Write(text)
+ }
+ fmt.Fprintf(&buf, "%s</%s>\n", prefix, b.Key)
+
+ return buf.Bytes(), nil
+}
+
+func valuesMarshalText(values []Value) (string, error) {
+ var b strings.Builder
+
+ for _, v := range values {
+ switch v := v.Interface().(type) {
+ case string:
+ fmt.Fprintf(&b, " %q", v)
+ case float64, bool:
+ fmt.Fprintf(&b, " %v", v)
+ default:
+ return "", fmt.Errorf("unexpected value type: %T", v)
+ }
+ }
+
+ return b.String(), nil
+}
+
+// Port represents a port number in the configuration. When a configuration is
+// converted to Go types using Unmarshal, it implements special conversion
+// rules:
+// If the config option is a numeric value, it is ensured to be in the range
+// [1–65535]. If the config option is a string, it is converted to a port
+// number using "net".LookupPort (using "tcp" as network).
+type Port int
+
+// UnmarshalConfig converts b to a port number.
+func (p *Port) UnmarshalConfig(b Block) error {
+ if len(b.Values) != 1 || len(b.Children) != 0 {
+ return fmt.Errorf("option %q has to be a single scalar value", b.Key)
+ }
+
+ v := b.Values[0]
+ if f, ok := v.Float64(); ok {
+ if math.IsNaN(f) {
+ return fmt.Errorf("the value of the %q option (%v) is invalid", b.Key, f)
+ }
+ if f < 1 || f > math.MaxUint16 {
+ return fmt.Errorf("the value of the %q option (%v) is out of range", b.Key, f)
+ }
+ *p = Port(f)
+ return nil
+ }
+
+ if !v.IsString() {
+ return fmt.Errorf("the value of the %q option must be a number or a string", b.Key)
+ }
+
+ port, err := net.LookupPort("tcp", v.String())
+ if err != nil {
+ return fmt.Errorf("%s: %w", b.Key, err)
+ }
+
+ *p = Port(port)
+ return nil
+}
diff --git a/config/config_test.go b/config/config_test.go
new file mode 100644
index 0000000..3646b1b
--- /dev/null
+++ b/config/config_test.go
@@ -0,0 +1,629 @@
+package config
+
+import (
+ "fmt"
+ "math"
+ "testing"
+
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+)
+
+type dstConf2 struct {
+ Args string
+ KeepAlive bool
+ Expect []string
+ Hats int8
+}
+type dstConf struct {
+ Args string
+ Host dstConf2
+}
+type dstConf3 struct {
+ Args string
+ Host []dstConf2
+}
+
+// doubleInt implements the Unmarshaler interface to double the values on assignment.
+type doubleInt int
+
+func (di *doubleInt) UnmarshalConfig(block Block) error {
+ if len(block.Values) != 1 || len(block.Children) != 0 {
+ return fmt.Errorf("got %d values and %d children, want scalar value",
+ len(block.Values), len(block.Children))
+ }
+
+ n, ok := block.Values[0].Float64()
+ if !ok {
+ return fmt.Errorf("got a %T, want a number", block.Values[0].Interface())
+ }
+
+ *di = doubleInt(2.0 * n)
+ return nil
+}
+
+func stringPtr(s string) *string {
+ return &s
+}
+
+func TestConfig_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ src Block
+ dst interface{}
+ want interface{}
+ wantErr bool
+ }{
+ {
+ name: "Base test",
+ src: Block{
+ Key: "myPlugin",
+ Children: []Block{
+ {
+ Key: "Host",
+ Children: []Block{
+ {
+ Key: "KeepAlive",
+ Values: Values(true),
+ },
+ {
+ Key: "Expect",
+ Values: Values("foo"),
+ },
+ {
+ Key: "Expect",
+ Values: Values("bar"),
+ },
+ {
+ Key: "Hats",
+ Values: Values(424242.42),
+ },
+ },
+ },
+ },
+ },
+ dst: &dstConf{},
+ want: &dstConf{
+ Host: dstConf2{
+ KeepAlive: true,
+ Expect: []string{"foo", "bar"},
+ Hats: 50, // truncated to 8bit
+ },
+ },
+ },
+ {
+ name: "Test slice of struct",
+ src: Block{
+ Key: "myPlugin",
+ Children: []Block{
+ {
+ Key: "Host",
+ Children: []Block{
+ {
+ Key: "KeepAlive",
+ Values: Values(true),
+ },
+ {
+ Key: "Expect",
+ Values: Values("foo"),
+ },
+ {
+ Key: "Expect",
+ Values: Values("bar"),
+ },
+ {
+ Key: "Hats",
+ Values: Values(424242.42),
+ },
+ },
+ },
+ },
+ },
+ dst: &dstConf3{},
+ want: &dstConf3{
+ Host: []dstConf2{
+ {
+ KeepAlive: true,
+ Expect: []string{"foo", "bar"},
+ Hats: 50, // truncated to 8bit
+ },
+ },
+ },
+ },
+ {
+ name: "nil argument",
+ dst: nil,
+ wantErr: true,
+ },
+ {
+ name: "non-pointer argument",
+ dst: int(23),
+ wantErr: true,
+ },
+ {
+ name: "block values",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ },
+ dst: &dstConf{},
+ want: &dstConf{
+ Args: "test",
+ },
+ },
+ {
+ name: "multiple block values",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("one", "two"),
+ },
+ dst: &struct {
+ Args []string
+ }{},
+ want: &struct {
+ Args []string
+ }{
+ Args: []string{"one", "two"},
+ },
+ },
+ {
+ name: "block values but no Args field",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ },
+ dst: &struct{}{},
+ wantErr: true,
+ },
+ {
+ name: "block values with type mismatch",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("not an int"),
+ },
+ dst: &struct {
+ Args []int
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "multiple block values but scalar Args field",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("one", "two"),
+ },
+ dst: &dstConf{},
+ wantErr: true,
+ },
+ {
+ name: "block with children requires struct",
+ src: Block{
+ Key: "Plugin",
+ Children: make([]Block, 1),
+ },
+ dst: stringPtr("not a struct"),
+ wantErr: true,
+ },
+ {
+ name: "error in nested block",
+ src: Block{
+ Key: "Plugin",
+ Children: []Block{
+ {
+ Key: "BlockWithErrors",
+ Values: Values("have string, expect int"),
+ Children: make([]Block, 1),
+ },
+ },
+ },
+ dst: &struct {
+ BlockWithErrors struct {
+ Args int // type mismatch
+ }
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "unexpected nested block",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ Children: []Block{
+ {
+ Key: "UnexpectedBlock",
+ Children: make([]Block, 1),
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "unmarshal list into scalar fails",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ Children: []Block{
+ {
+ Key: "ListValue",
+ Values: Values(23, 64),
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ ListValue float64
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "unmarshal into channel fails",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ Children: []Block{
+ {
+ Key: "NumberValue",
+ Values: Values(64),
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ NumberValue chan struct{}
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "unmarshal interface success",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ Children: []Block{
+ {
+ Key: "Double",
+ Values: Values(64),
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Double doubleInt
+ }{},
+ want: &struct {
+ Args string
+ Double doubleInt
+ }{
+ Args: "test",
+ Double: doubleInt(128),
+ },
+ },
+ {
+ name: "unmarshal interface failure",
+ src: Block{
+ Key: "Plugin",
+ Values: Values("test"),
+ Children: []Block{
+ {
+ Key: "Double",
+ Values: Values("not a number", 64),
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Double doubleInt
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "port numeric success",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{Float64(80)},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ want: &struct {
+ Args string
+ Port Port
+ }{
+ Args: "test",
+ Port: Port(80),
+ },
+ },
+ {
+ name: "port out of range",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{Float64(1 << 48)},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "port not a number",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{Float64(math.NaN())},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "port invalid type",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{Bool(true)},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ wantErr: true,
+ },
+ {
+ name: "port string success",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{String("http")},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ want: &struct {
+ Args string
+ Port Port
+ }{
+ Args: "test",
+ Port: Port(80),
+ },
+ },
+ {
+ name: "port string failure",
+ src: Block{
+ Key: "Plugin",
+ Values: []Value{String("test")},
+ Children: []Block{
+ {
+ Key: "Port",
+ Values: []Value{String("--- invalid ---")},
+ },
+ },
+ },
+ dst: &struct {
+ Args string
+ Port Port
+ }{},
+ wantErr: true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if err := tt.src.Unmarshal(tt.dst); (err != nil) != tt.wantErr {
+ t.Errorf("Unmarshal() = %v, wantErr %v", err, tt.wantErr)
+ }
+ if tt.wantErr {
+ return
+ }
+ if diff := cmp.Diff(tt.want, tt.dst); diff != "" {
+ t.Errorf("%#v.Unmarshal() result differs (+got/-want):\n%s", tt.src, diff)
+ }
+ })
+ }
+}
+
+func TestValues(t *testing.T) {
+ cases := []struct {
+ in interface{}
+ want Value
+ }{
+ // exact matches
+ {nil, Float64(math.NaN())},
+ {"foo", String("foo")},
+ {[]byte("byte array"), String("byte array")},
+ {true, Bool(true)},
+ {float64(42.11), Float64(42.11)},
+ // convertible to float64
+ {float32(12.25), Float64(12.25)},
+ {int(0x1F622), Float64(128546)},
+ {uint64(0x1F61F), Float64(128543)},
+ // not convertiable to float64
+ {complex(4, 1), String("(4+1i)")},
+ {struct{}{}, String("{}")},
+ {map[string]int{"answer": 42}, String("map[answer:42]")},
+ {[]int{1, 2, 3}, String("[1 2 3]")},
+ }
+
+ opts := []cmp.Option{
+ cmp.AllowUnexported(Value{}),
+ cmpopts.EquateNaNs(),
+ }
+ for _, tc := range cases {
+ got := Values(tc.in)
+ want := []Value{tc.want}
+
+ if !cmp.Equal(want, got, opts...) {
+ t.Errorf("Values(%#v) = %v, want %v", tc.in, got, want)
+ }
+ }
+}
+
+func TestValue_Interface(t *testing.T) {
+ cases := []struct {
+ v Value
+ want interface{}
+ }{
+ {String("foo"), "foo"},
+ {Float64(42.0), 42.0},
+ {Bool(true), true},
+ {Value{}, ""}, // zero value is a string
+ }
+
+ for _, tc := range cases {
+ got := tc.v.Interface()
+ if !cmp.Equal(tc.want, got) {
+ t.Errorf("%#v.Interface() = %v, want %v", tc.v, got, tc.want)
+ }
+ }
+}
+
+func TestBlock_Merge(t *testing.T) {
+ makeBlock := func(key, value string, children []Block) Block {
+ return Block{
+ Key: key,
+ Values: Values(value),
+ Children: children,
+ }
+ }
+
+ makeChildren := func(names ...string) []Block {
+ var ret []Block
+ for _, n := range names {
+ ret = append(ret, makeBlock(n, "value", nil))
+ }
+ return ret
+ }
+
+ cases := []struct {
+ name string
+ in0, in1 Block
+ want Block
+ wantErr bool
+ }{
+ {
+ name: "success",
+ in0: makeBlock("Plugin", "test", makeChildren("foo")),
+ in1: makeBlock("Plugin", "test", makeChildren("bar")),
+ want: makeBlock("Plugin", "test", makeChildren("foo", "bar")),
+ },
+ {
+ name: "destination without children",
+ in0: makeBlock("Plugin", "test", nil),
+ in1: makeBlock("Plugin", "test", makeChildren("bar")),
+ want: makeBlock("Plugin", "test", makeChildren("bar")),
+ },
+ {
+ name: "source without children",
+ in0: makeBlock("Plugin", "test", makeChildren("foo")),
+ in1: makeBlock("Plugin", "test", nil),
+ want: makeBlock("Plugin", "test", makeChildren("foo")),
+ },
+ {
+ name: "source and destination without children",
+ in0: makeBlock("Plugin", "test", nil),
+ in1: makeBlock("Plugin", "test", nil),
+ want: makeBlock("Plugin", "test", nil),
+ },
+ {
+ name: "merge into zero value",
+ in0: Block{},
+ in1: makeBlock("Plugin", "test", makeChildren("foo")),
+ want: makeBlock("Plugin", "test", makeChildren("foo")),
+ },
+ {
+ name: "key mismatch",
+ in0: makeBlock("Plugin", "test", nil),
+ in1: makeBlock("SomethingElse", "test", nil),
+ wantErr: true,
+ },
+ {
+ name: "value mismatch",
+ in0: makeBlock("Plugin", "test", nil),
+ in1: makeBlock("Plugin", "prod", nil),
+ wantErr: true,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Logf("block = %#v", tc.in0)
+ err := tc.in0.Merge(tc.in1)
+ if gotErr := err != nil; gotErr != tc.wantErr {
+ t.Errorf("block.Merge() = %v, want error %v", err, tc.wantErr)
+ }
+ if tc.wantErr {
+ return
+ }
+
+ if diff := cmp.Diff(tc.want, tc.in0, cmp.AllowUnexported(Value{})); diff != "" {
+ t.Errorf("other block = %#v", tc.in1)
+ t.Errorf("block.Merge() differd (+got/-want)\n%s", diff)
+ }
+ })
+ }
+}
+
+func TestBlock_MarshalText(t *testing.T) {
+ b := Block{
+ Key: "myPlugin",
+ Children: []Block{
+ {
+ Key: "Listen",
+ Values: Values("localhost", 8080),
+ Children: []Block{
+ {
+ Key: "KeepAlive",
+ Values: Values(true),
+ },
+ },
+ },
+ },
+ }
+
+ data, err := b.MarshalText()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ want := `<myPlugin>
+ <Listen "localhost" 8080>
+ KeepAlive true
+ </Listen>
+</myPlugin>
+`
+ if diff := cmp.Diff(want, string(data)); diff != "" {
+ t.Errorf("MarshalText() differs (-got/+want):\n%s", diff)
+ }
+}
diff --git a/debian/changelog b/debian/changelog
index a7c7a6d..58e188d 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,9 @@
+golang-collectd (0.5.0+git20200608.1.92e86f9-1) UNRELEASED; urgency=low
+
+ * New upstream snapshot.
+
+ -- Debian Janitor <janitor@jelmer.uk> Sat, 16 Apr 2022 22:43:26 -0000
+
golang-collectd (0.3.0+git20181025.f80706d-2) unstable; urgency=medium
* Remove myself from uploaders.
diff --git a/exec/exec.go b/exec/exec.go
index 5544b90..956c28e 100644
--- a/exec/exec.go
+++ b/exec/exec.go
@@ -15,19 +15,7 @@ import (
)
// Putval is the dispatcher used by the exec package to print ValueLists.
-var Putval = format.NewPutval(os.Stdout)
-
-type valueCallback struct {
- callback func() api.Value
- vl *api.ValueList
- done chan bool
-}
-
-type voidCallback struct {
- callback func(context.Context, time.Duration)
- interval time.Duration
- done chan bool
-}
+var Putval api.Writer = format.NewPutval(os.Stdout)
type callback interface {
run(context.Context, *sync.WaitGroup)
@@ -51,10 +39,10 @@ func NewExecutor() *Executor {
// only returns a Number, i.e. either a api.Gauge or api.Derive, and formatting
// and printing is done by the executor.
func (e *Executor) ValueCallback(callback func() api.Value, vl *api.ValueList) {
- e.cb = append(e.cb, valueCallback{
+ e.cb = append(e.cb, &valueCallback{
callback: callback,
- vl: vl,
- done: make(chan bool),
+ vl: *vl,
+ done: make(chan struct{}),
})
}
@@ -67,7 +55,7 @@ func (e *Executor) VoidCallback(callback func(context.Context, time.Duration), i
e.cb = append(e.cb, voidCallback{
callback: callback,
interval: interval,
- done: make(chan bool),
+ done: make(chan struct{}),
})
}
@@ -89,48 +77,64 @@ func (e *Executor) Stop() {
}
}
-func (cb valueCallback) run(ctx context.Context, g *sync.WaitGroup) {
+type valueCallback struct {
+ callback func() api.Value
+ vl api.ValueList
+ done chan struct{}
+}
+
+func (cb *valueCallback) run(ctx context.Context, g *sync.WaitGroup) {
+ defer g.Done()
+
if cb.vl.Host == "" {
cb.vl.Host = Hostname()
}
cb.vl.Interval = sanitizeInterval(cb.vl.Interval)
- cb.vl.Values = make([]api.Value, 1)
ticker := time.NewTicker(cb.vl.Interval)
-
for {
select {
- case _ = <-ticker.C:
- cb.vl.Values[0] = cb.callback()
+ case <-ticker.C:
+ cb.vl.Values = []api.Value{cb.callback()}
cb.vl.Time = time.Now()
- Putval.Write(ctx, cb.vl)
- case _ = <-cb.done:
- g.Done()
+ Putval.Write(ctx, &cb.vl)
+ case <-cb.done:
+ return
+ case <-ctx.Done():
return
}
}
}
-func (cb valueCallback) stop() {
- cb.done <- true
+func (cb *valueCallback) stop() {
+ close(cb.done)
+}
+
+type voidCallback struct {
+ callback func(context.Context, time.Duration)
+ interval time.Duration
+ done chan struct{}
}
func (cb voidCallback) run(ctx context.Context, g *sync.WaitGroup) {
+ defer g.Done()
+
ticker := time.NewTicker(sanitizeInterval(cb.interval))
for {
select {
- case _ = <-ticker.C:
+ case <-ticker.C:
cb.callback(ctx, cb.interval)
- case _ = <-cb.done:
- g.Done()
+ case <-cb.done:
+ return
+ case <-ctx.Done():
return
}
}
}
func (cb voidCallback) stop() {
- cb.done <- true
+ close(cb.done)
}
// Interval determines the default interval from the "COLLECTD_INTERVAL"
diff --git a/exec/exec_test.go b/exec/exec_test.go
index a13125b..ed8189c 100644
--- a/exec/exec_test.go
+++ b/exec/exec_test.go
@@ -10,32 +10,34 @@ import (
)
func TestSanitizeInterval(t *testing.T) {
- var got, want time.Duration
-
- got = sanitizeInterval(10 * time.Second)
- want = 10 * time.Second
- if got != want {
- t.Errorf("got %v, want %v", got, want)
+ cases := []struct {
+ arg time.Duration
+ env string
+ want time.Duration
+ }{
+ {42 * time.Second, "", 42 * time.Second},
+ {42 * time.Second, "23", 42 * time.Second},
+ {0, "23", 23 * time.Second},
+ {0, "8.15", 8150 * time.Millisecond},
+ {0, "", 10 * time.Second},
+ {0, "--- INVALID ---", 10 * time.Second},
}
- // Environment with seconds
- if err := os.Setenv("COLLECTD_INTERVAL", "42"); err != nil {
- t.Fatalf("os.Setenv: %v", err)
- }
- got = sanitizeInterval(0)
- want = 42 * time.Second
- if got != want {
- t.Errorf("got %v, want %v", got, want)
- }
+ for _, tc := range cases {
+ if tc.env != "" {
+ if err := os.Setenv("COLLECTD_INTERVAL", tc.env); err != nil {
+ t.Fatal(err)
+ }
+ } else { // tc.env == ""
+ if err := os.Unsetenv("COLLECTD_INTERVAL"); err != nil {
+ t.Fatal(err)
+ }
+ }
- // Environment with milliseconds
- if err := os.Setenv("COLLECTD_INTERVAL", "31.337"); err != nil {
- t.Fatalf("os.Setenv: %v", err)
- }
- got = sanitizeInterval(0)
- want = 31337 * time.Millisecond
- if got != want {
- t.Errorf("got %v, want %v", got, want)
+ got := sanitizeInterval(tc.arg)
+ if got != tc.want {
+ t.Errorf("COLLECTD_INTERVAL=%q sanitizeInterval(%v) = %v, want %v", tc.env, tc.arg, got, tc.want)
+ }
}
}
diff --git a/exec/exec_x_test.go b/exec/exec_x_test.go
new file mode 100644
index 0000000..e129774
--- /dev/null
+++ b/exec/exec_x_test.go
@@ -0,0 +1,150 @@
+package exec_test
+
+import (
+ "context"
+ "errors"
+ "os"
+ "sync"
+ "testing"
+ "time"
+
+ "collectd.org/api"
+ "collectd.org/exec"
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+)
+
+type testWriter struct {
+ vl *api.ValueList
+}
+
+func (w *testWriter) Write(_ context.Context, vl *api.ValueList) error {
+ if w.vl != nil {
+ return errors.New("received unexpected second value")
+ }
+
+ w.vl = vl
+ return nil
+}
+
+func TestValueCallback_ExecutorStop(t *testing.T) {
+ cases := []struct {
+ title string
+ stopFunc func(f context.CancelFunc, e *exec.Executor)
+ }{
+ {"ExecutorStop", func(_ context.CancelFunc, e *exec.Executor) { e.Stop() }},
+ {"CancelContext", func(cancel context.CancelFunc, _ *exec.Executor) { cancel() }},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ if err := os.Setenv("COLLECTD_HOSTNAME", "example.com"); err != nil {
+ t.Fatal(err)
+ }
+ defer func() {
+ os.Unsetenv("COLLECTD_HOSTNAME")
+ }()
+
+ savedPutval := exec.Putval
+ defer func() {
+ exec.Putval = savedPutval
+ }()
+
+ w := &testWriter{}
+ exec.Putval = w
+
+ e := exec.NewExecutor()
+ ch := make(chan struct{})
+ go func() {
+ // wait for ch to be closed
+ <-ch
+ tc.stopFunc(cancel, e)
+ }()
+
+ var once sync.Once
+ e.ValueCallback(func() api.Value {
+ once.Do(func() {
+ close(ch)
+ })
+ return api.Derive(42)
+ }, &api.ValueList{
+ Identifier: api.Identifier{
+ Plugin: "go-exec",
+ Type: "derive",
+ },
+ Interval: time.Millisecond,
+ DSNames: []string{"value"},
+ })
+
+ // e.Run() blocks until the context is canceled or
+ // e.Stop() is called (see tc.stopFunc above).
+ e.Run(ctx)
+
+ want := &api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "go-exec",
+ Type: "derive",
+ },
+ Interval: time.Millisecond,
+ Values: []api.Value{api.Derive(42)},
+ DSNames: []string{"value"},
+ }
+ if diff := cmp.Diff(want, w.vl, cmpopts.IgnoreFields(api.ValueList{}, "Time")); diff != "" {
+ t.Errorf("received value lists differ (+got/-want):\n%s", diff)
+ }
+ })
+ }
+}
+
+func TestVoidCallback(t *testing.T) {
+ cases := []struct {
+ title string
+ stopFunc func(f context.CancelFunc, e *exec.Executor)
+ }{
+ {"ExecutorStop", func(_ context.CancelFunc, e *exec.Executor) { e.Stop() }},
+ {"CancelContext", func(cancel context.CancelFunc, _ *exec.Executor) { cancel() }},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ e := exec.NewExecutor()
+ ch := make(chan struct{})
+ go func() {
+ // wait for ch to be closed
+ <-ch
+ tc.stopFunc(cancel, e)
+ }()
+
+ var (
+ calls int
+ once sync.Once
+ )
+ e.VoidCallback(func(_ context.Context, d time.Duration) {
+ if got, want := d, time.Millisecond; got != want {
+ t.Errorf("VoidCallback(%v), want argument %v", got, want)
+ }
+
+ calls++
+
+ once.Do(func() {
+ close(ch)
+ })
+ }, time.Millisecond)
+
+ // e.Run() blocks until the context is canceled or
+ // e.Stop() is called (see tc.stopFunc above).
+ e.Run(ctx)
+
+ if got, want := calls, 1; got != want {
+ t.Errorf("number of calls = %d, want %d", got, want)
+ }
+ })
+ }
+}
diff --git a/export/export.go b/export/export.go
index 2a88f30..b56d54e 100644
--- a/export/export.go
+++ b/export/export.go
@@ -19,13 +19,16 @@ Gauge.
// Call Run() in its own goroutine.
func main() {
+ ctx := context.Background()
+
+ // Any other type implementing api.Writer works.
client, err := network.Dial(
net.JoinHostPort(network.DefaultIPv6Address, network.DefaultService),
network.ClientOptions{})
if err != nil {
log.Fatal(err)
- }
- go export.Run(client, export.Options{
+ }
+ go export.Run(ctx, client, export.Options{
Interval: 10 * time.Second,
})
// …
@@ -75,24 +78,26 @@ type Options struct {
}
// Run periodically calls the ValueList function of each Var, sets the Time and
-// Interval fields and passes it w.Write(). This function blocks indefinitely.
+// Interval fields and passes it w.Write(). This function blocks until the
+// context is cancelled.
func Run(ctx context.Context, w api.Writer, opts Options) error {
ticker := time.NewTicker(opts.Interval)
for {
select {
- case _ = <-ticker.C:
+ case <-ticker.C:
mutex.RLock()
for _, v := range vars {
vl := v.ValueList()
vl.Time = time.Now()
vl.Interval = opts.Interval
if err := w.Write(ctx, vl); err != nil {
- mutex.RUnlock()
- return err
+ log.Printf("%T.Write(): %v", w, err)
}
}
mutex.RUnlock()
+ case <-ctx.Done():
+ return ctx.Err()
}
}
}
diff --git a/export/export_test.go b/export/export_test.go
index 52a22b5..b0707b1 100644
--- a/export/export_test.go
+++ b/export/export_test.go
@@ -1,20 +1,25 @@
package export // import "collectd.org/export"
import (
+ "context"
+ "errors"
"expvar"
- "reflect"
+ "sync"
"testing"
+ "time"
"collectd.org/api"
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
)
func TestDerive(t *testing.T) {
- d := NewDerive(api.Identifier{
- Host: "example.com",
- Plugin: "golang",
- Type: "derive",
- })
+ // clean up shared resource after testing
+ defer func() {
+ vars = nil
+ }()
+ d := NewDeriveString("example.com/TestDerive/derive")
for i := 0; i < 10; i++ {
d.Add(i)
}
@@ -22,48 +27,136 @@ func TestDerive(t *testing.T) {
want := &api.ValueList{
Identifier: api.Identifier{
Host: "example.com",
- Plugin: "golang",
+ Plugin: "TestDerive",
Type: "derive",
},
Values: []api.Value{api.Derive(45)},
}
got := d.ValueList()
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %#v, want %#v", got, want)
+ if diff := cmp.Diff(want, got); diff != "" {
+ t.Errorf("Derive.ValueList() differs (+got/-want):\n%s", diff)
}
- s := expvar.Get("example.com/golang/derive").String()
+ s := expvar.Get("example.com/TestDerive/derive").String()
if s != "45" {
t.Errorf("got %q, want %q", s, "45")
}
}
func TestGauge(t *testing.T) {
- g := NewGauge(api.Identifier{
- Host: "example.com",
- Plugin: "golang",
- Type: "gauge",
- })
+ // clean up shared resource after testing
+ defer func() {
+ vars = nil
+ }()
+ g := NewGaugeString("example.com/TestGauge/gauge")
g.Set(42.0)
want := &api.ValueList{
Identifier: api.Identifier{
Host: "example.com",
- Plugin: "golang",
+ Plugin: "TestGauge",
Type: "gauge",
},
Values: []api.Value{api.Gauge(42)},
}
got := g.ValueList()
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %#v, want %#v", got, want)
+ if diff := cmp.Diff(want, got); diff != "" {
+ t.Errorf("Gauge.ValueList() differs (+got/-want):\n%s", diff)
}
- s := expvar.Get("example.com/golang/gauge").String()
+ s := expvar.Get("example.com/TestGauge/gauge").String()
if s != "42" {
t.Errorf("got %q, want %q", s, "42")
}
}
+
+type testWriter struct {
+ got []*api.ValueList
+ done chan<- struct{}
+ once *sync.Once
+}
+
+func (w *testWriter) Write(ctx context.Context, vl *api.ValueList) error {
+ w.got = append(w.got, vl)
+ w.once.Do(func() {
+ close(w.done)
+ })
+ return nil
+}
+
+func TestRun(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ // clean up shared resource after testing
+ defer func() {
+ vars = nil
+ }()
+
+ d := NewDeriveString("example.com/TestRun/derive")
+ d.Add(23)
+
+ g := NewGaugeString("example.com/TestRun/gauge")
+ g.Set(42)
+
+ var (
+ done = make(chan struct{})
+ once sync.Once
+ )
+
+ w := testWriter{
+ done: done,
+ once: &once,
+ }
+
+ go func() {
+ // when one metric has been written, cancel the context
+ <-done
+ cancel()
+ }()
+
+ err := Run(ctx, &w, Options{Interval: 100 * time.Millisecond})
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("Run() = %v, want %v", err, context.Canceled)
+ }
+
+ want := []*api.ValueList{
+ {
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestRun",
+ Type: "gauge",
+ },
+ Time: time.Now(),
+ Interval: 100 * time.Millisecond,
+ Values: []api.Value{api.Gauge(42)},
+ },
+ {
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestRun",
+ Type: "derive",
+ },
+ Time: time.Now(),
+ Interval: 100 * time.Millisecond,
+ Values: []api.Value{api.Derive(23)},
+ },
+ }
+
+ ignoreOrder := cmpopts.SortSlices(func(a, b *api.ValueList) bool {
+ return a.Identifier.String() < b.Identifier.String()
+ })
+ approximateTime := cmp.Comparer(func(t0, t1 time.Time) bool {
+ diff := t0.Sub(t1)
+ if t1.After(t0) {
+ diff = t1.Sub(t0)
+ }
+
+ return diff < 2*time.Second
+ })
+ if diff := cmp.Diff(want, w.got, ignoreOrder, approximateTime); diff != "" {
+ t.Errorf("received value lists differ (+got/-want):\n%s", diff)
+ }
+}
diff --git a/format/putval.go b/format/putval.go
index cbe20e9..61bcc25 100644
--- a/format/putval.go
+++ b/format/putval.go
@@ -6,10 +6,13 @@ import (
"context"
"fmt"
"io"
+ "log"
"strings"
+ "sync"
"time"
"collectd.org/api"
+ "collectd.org/meta"
)
// Putval implements the Writer interface for PUTVAL formatted output.
@@ -32,8 +35,8 @@ func (p *Putval) Write(_ context.Context, vl *api.ValueList) error {
return err
}
- _, err = fmt.Fprintf(p.w, "PUTVAL %q interval=%.3f %s\n",
- vl.Identifier.String(), vl.Interval.Seconds(), s)
+ _, err = fmt.Fprintf(p.w, "PUTVAL %q interval=%.3f %s%s\n",
+ vl.Identifier.String(), vl.Interval.Seconds(), formatMeta(vl.Meta), s)
return err
}
@@ -65,3 +68,25 @@ func formatTime(t time.Time) string {
return fmt.Sprintf("%.3f", float64(t.UnixNano())/1000000000.0)
}
+
+var stringWarning sync.Once
+
+func formatMeta(m meta.Data) string {
+ if len(m) == 0 {
+ return ""
+ }
+
+ var values []string
+ for k, v := range m {
+ // collectd only supports string meta data values as of 5.11.
+ if !v.IsString() {
+ stringWarning.Do(func() {
+ log.Printf("Non-string metadata not supported yet")
+ })
+ continue
+ }
+ values = append(values, fmt.Sprintf("meta:%s=%q ", k, v.String()))
+ }
+
+ return strings.Join(values, "")
+}
diff --git a/format/putval_test.go b/format/putval_test.go
new file mode 100644
index 0000000..4169a42
--- /dev/null
+++ b/format/putval_test.go
@@ -0,0 +1,118 @@
+package format_test
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "collectd.org/api"
+ "collectd.org/format"
+ "collectd.org/meta"
+ "github.com/google/go-cmp/cmp"
+)
+
+func TestPutval(t *testing.T) {
+ baseVL := api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestPutval",
+ Type: "derive",
+ },
+ Interval: 10 * time.Second,
+ Values: []api.Value{api.Derive(42)},
+ DSNames: []string{"value"},
+ }
+
+ cases := []struct {
+ title string
+ modify func(*api.ValueList)
+ want string
+ wantErr bool
+ }{
+ {
+ title: "derive",
+ want: `PUTVAL "example.com/TestPutval/derive" interval=10.000 N:42` + "\n",
+ },
+ {
+ title: "gauge",
+ modify: func(vl *api.ValueList) {
+ vl.Type = "gauge"
+ vl.Values = []api.Value{api.Gauge(20.0 / 3.0)}
+ },
+ want: `PUTVAL "example.com/TestPutval/gauge" interval=10.000 N:6.66666666666667` + "\n",
+ },
+ {
+ title: "counter",
+ modify: func(vl *api.ValueList) {
+ vl.Type = "counter"
+ vl.Values = []api.Value{api.Counter(31337)}
+ },
+ want: `PUTVAL "example.com/TestPutval/counter" interval=10.000 N:31337` + "\n",
+ },
+ {
+ title: "multiple values",
+ modify: func(vl *api.ValueList) {
+ vl.Type = "if_octets"
+ vl.Values = []api.Value{api.Derive(1), api.Derive(2)}
+ vl.DSNames = []string{"rx", "tx"}
+ },
+ want: `PUTVAL "example.com/TestPutval/if_octets" interval=10.000 N:1:2` + "\n",
+ },
+ {
+ title: "invalid type",
+ modify: func(vl *api.ValueList) {
+ vl.Values = []api.Value{nil}
+ },
+ wantErr: true,
+ },
+ {
+ title: "time",
+ modify: func(vl *api.ValueList) {
+ vl.Time = time.Unix(1588087972, 987654321)
+ },
+ want: `PUTVAL "example.com/TestPutval/derive" interval=10.000 1588087972.988:42` + "\n",
+ },
+ {
+ title: "interval",
+ modify: func(vl *api.ValueList) {
+ vl.Interval = 9876543 * time.Microsecond
+ },
+ want: `PUTVAL "example.com/TestPutval/derive" interval=9.877 N:42` + "\n",
+ },
+ {
+ title: "meta_data",
+ modify: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.String("value"),
+ "ignored": meta.Bool(true),
+ }
+ },
+ want: `PUTVAL "example.com/TestPutval/derive" interval=10.000 meta:key="value" N:42` + "\n",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ ctx := context.Background()
+
+ vl := baseVL
+ if tc.modify != nil {
+ tc.modify(&vl)
+ }
+
+ var b strings.Builder
+ err := format.NewPutval(&b).Write(ctx, &vl)
+ if gotErr := err != nil; gotErr != tc.wantErr {
+ t.Errorf("Putval.Write(%#v) = %v, want error %v", &vl, err, tc.wantErr)
+ }
+ if tc.wantErr {
+ return
+ }
+
+ if diff := cmp.Diff(tc.want, b.String()); diff != "" {
+ t.Errorf("Putval.Write(%#v) differs (+got/-want):\n%s", &vl, diff)
+ }
+ })
+ }
+}
diff --git a/meta/meta.go b/meta/meta.go
new file mode 100644
index 0000000..0944983
--- /dev/null
+++ b/meta/meta.go
@@ -0,0 +1,181 @@
+// Package meta provides data types for collectd meta data.
+//
+// Meta data can be associated with value lists (api.ValueList) and
+// notifications (not yet implemented in the collectd Go API).
+package meta
+
+import (
+ "encoding/json"
+ "fmt"
+ "math"
+)
+
+type entryType int
+
+const (
+ _ entryType = iota
+ metaStringType
+ metaInt64Type
+ metaUInt64Type
+ metaFloat64Type
+ metaBoolType
+)
+
+// Data is a map of meta data values. No setter and getter methods are
+// implemented for this, callers are expected to add and remove entries as they
+// would from a normal map.
+type Data map[string]Entry
+
+// Clone returns a copy of d.
+func (d Data) Clone() Data {
+ if d == nil {
+ return nil
+ }
+
+ cpy := make(Data)
+ for k, v := range d {
+ cpy[k] = v
+ }
+ return cpy
+}
+
+// Entry is an entry in the metadata set. The typed value may be bool, float64,
+// int64, uint64, or string.
+type Entry struct {
+ s string
+ i int64
+ u uint64
+ f float64
+ b bool
+
+ typ entryType
+}
+
+// Bool returns a new bool Entry.
+func Bool(b bool) Entry { return Entry{b: b, typ: metaBoolType} }
+
+// Float64 returns a new float64 Entry.
+func Float64(f float64) Entry { return Entry{f: f, typ: metaFloat64Type} }
+
+// Int64 returns a new int64 Entry.
+func Int64(i int64) Entry { return Entry{i: i, typ: metaInt64Type} }
+
+// UInt64 returns a new uint64 Entry.
+func UInt64(u uint64) Entry { return Entry{u: u, typ: metaUInt64Type} }
+
+// String returns a new string Entry.
+func String(s string) Entry { return Entry{s: s, typ: metaStringType} }
+
+// Bool returns the bool value of e.
+func (e Entry) Bool() (value, ok bool) { return e.b, e.typ == metaBoolType }
+
+// Float64 returns the float64 value of e.
+func (e Entry) Float64() (float64, bool) { return e.f, e.typ == metaFloat64Type }
+
+// Int64 returns the int64 value of e.
+func (e Entry) Int64() (int64, bool) { return e.i, e.typ == metaInt64Type }
+
+// UInt64 returns the uint64 value of e.
+func (e Entry) UInt64() (uint64, bool) { return e.u, e.typ == metaUInt64Type }
+
+// String returns a string representation of e.
+func (e Entry) String() string {
+ switch e.typ {
+ case metaBoolType:
+ return fmt.Sprintf("%v", e.b)
+ case metaFloat64Type:
+ return fmt.Sprintf("%.15g", e.f)
+ case metaInt64Type:
+ return fmt.Sprintf("%v", e.i)
+ case metaUInt64Type:
+ return fmt.Sprintf("%v", e.u)
+ case metaStringType:
+ return e.s
+ default:
+ return fmt.Sprintf("%v", nil)
+ }
+}
+
+// IsString returns true if e is a string value.
+func (e Entry) IsString() bool {
+ return e.typ == metaStringType
+}
+
+// Interface returns e's value. It is intended to be used with type switches
+// and when printing an entry's type with the "%T" formatting.
+func (e Entry) Interface() interface{} {
+ switch e.typ {
+ case metaBoolType:
+ return e.b
+ case metaFloat64Type:
+ return e.f
+ case metaInt64Type:
+ return e.i
+ case metaUInt64Type:
+ return e.u
+ case metaStringType:
+ return e.s
+ default:
+ return nil
+ }
+}
+
+// MarshalJSON implements the "encoding/json".Marshaller interface.
+func (e Entry) MarshalJSON() ([]byte, error) {
+ switch e.typ {
+ case metaBoolType:
+ return json.Marshal(e.b)
+ case metaFloat64Type:
+ if math.IsNaN(e.f) {
+ return json.Marshal(nil)
+ }
+ return json.Marshal(e.f)
+ case metaInt64Type:
+ return json.Marshal(e.i)
+ case metaUInt64Type:
+ return json.Marshal(e.u)
+ case metaStringType:
+ return json.Marshal(e.s)
+ default:
+ return json.Marshal(nil)
+ }
+}
+
+// UnmarshalJSON implements the "encoding/json".Unmarshaller interface.
+func (e *Entry) UnmarshalJSON(raw []byte) error {
+ var b *bool
+ if json.Unmarshal(raw, &b) == nil && b != nil {
+ *e = Bool(*b)
+ return nil
+ }
+
+ var s *string
+ if json.Unmarshal(raw, &s) == nil && s != nil {
+ *e = String(*s)
+ return nil
+ }
+
+ var i *int64
+ if json.Unmarshal(raw, &i) == nil && i != nil {
+ *e = Int64(*i)
+ return nil
+ }
+
+ var u *uint64
+ if json.Unmarshal(raw, &u) == nil && u != nil {
+ *e = UInt64(*u)
+ return nil
+ }
+
+ var f *float64
+ if json.Unmarshal(raw, &f) == nil {
+ if f != nil {
+ *e = Float64(*f)
+ } else {
+ *e = Float64(math.NaN())
+ }
+ return nil
+ }
+
+ return fmt.Errorf("unable to parse %q as meta entry", raw)
+}
diff --git a/meta/meta_test.go b/meta/meta_test.go
new file mode 100644
index 0000000..045b4bb
--- /dev/null
+++ b/meta/meta_test.go
@@ -0,0 +1,343 @@
+package meta_test
+
+import (
+ "encoding/json"
+ "fmt"
+ "log"
+ "math"
+ "math/rand"
+ "sort"
+ "testing"
+ "time"
+
+ "collectd.org/meta"
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+)
+
+func ExampleData() {
+ // Allocate new meta.Data object.
+ m := meta.Data{
+ // Add interger named "answer":
+ "answer": meta.Int64(42),
+ // Add bool named "panic":
+ "panic": meta.Bool(false),
+ }
+
+ // Add string named "required":
+ m["required"] = meta.String("towel")
+
+ // Remove the "panic" value:
+ delete(m, "panic")
+}
+
+func ExampleData_exists() {
+ m := meta.Data{
+ "answer": meta.Int64(42),
+ "panic": meta.Bool(false),
+ "required": meta.String("towel"),
+ }
+
+ for _, k := range []string{"answer", "question"} {
+ _, ok := m[k]
+ fmt.Println(k, "exists:", ok)
+ }
+
+ // Output:
+ // answer exists: true
+ // question exists: false
+}
+
+// This example demonstrates how to get a list of keys from meta.Data.
+func ExampleData_keys() {
+ m := meta.Data{
+ "answer": meta.Int64(42),
+ "panic": meta.Bool(false),
+ "required": meta.String("towel"),
+ }
+
+ var keys []string
+ for k := range m {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ fmt.Println(keys)
+
+ // Output:
+ // [answer panic required]
+}
+
+func ExampleEntry() {
+ // Allocate an int64 Entry.
+ answer := meta.Int64(42)
+
+ // Read back the "answer" value and ensure it is in fact an int64.
+ a, ok := answer.Int64()
+ if !ok {
+ log.Fatal("Answer is not an int64")
+ }
+ fmt.Printf("The answer is between %d and %d\n", a-1, a+1)
+
+ // Allocate a string Entry.
+ required := meta.String("towel")
+
+ // String is a bit different, because Entry.String() does not return a boolean.
+ // Check that "required" is a string and read it into a variable.
+ if !required.IsString() {
+ log.Fatal("required is not a string")
+ }
+ fmt.Println("You need a " + required.String())
+
+ // The fmt.Stringer interface is implemented for all value types. To
+ // print a string with default formatting, rely on the String() method:
+ p := meta.Bool(false)
+ fmt.Printf("Should I panic? %v\n", p)
+
+ // Output:
+ // The answer is between 41 and 43
+ // You need a towel
+ // Should I panic? false
+}
+
+func ExampleEntry_Interface() {
+ rand.Seed(time.Now().UnixNano())
+ m := meta.Data{}
+
+ // Create a value with unknown type. "key" is either a string,
+ // or an int64.
+ switch rand.Intn(2) {
+ case 0:
+ m["key"] = meta.String("value")
+ case 1:
+ m["key"] = meta.Int64(42)
+ }
+
+ // Scenario 0: A specific type is expected. Report an error that
+ // includes the actual type in the error message, if the value is of a
+ // different type.
+ if _, ok := m["key"].Int64(); !ok {
+ err := fmt.Errorf("key is a %T, want an int64", m["key"].Interface())
+ fmt.Println(err) // prints "key is a string, want an int64"
+ }
+
+ // Scenario 1: Multiple or all types need to be handled, for example to
+ // encode the meta data values. The most elegant solution for that is a
+ // type switch.
+ switch v := m["key"].Interface().(type) {
+ case string:
+ // string-specific code here
+ case int64:
+ // The above code skipped printing this, so print it here so
+ // this example produces the same output every time, despite
+ // the randomness.
+ fmt.Println("key is a string, want an int64")
+ default:
+ // Report the actual type if "key" is an unexpected type.
+ err := fmt.Errorf("unexpected type %T", v)
+ log.Fatal(err)
+ }
+
+ // Output:
+ // key is a string, want an int64
+}
+
+func TestMarshalJSON(t *testing.T) {
+ cases := []struct {
+ d meta.Data
+ want string
+ }{
+ {meta.Data{"foo": meta.Bool(true)}, `{"foo":true}`},
+ {meta.Data{"foo": meta.Float64(20.0 / 3.0)}, `{"foo":6.666666666666667}`},
+ {meta.Data{"foo": meta.Float64(math.NaN())}, `{"foo":null}`},
+ {meta.Data{"foo": meta.Int64(-42)}, `{"foo":-42}`},
+ {meta.Data{"foo": meta.UInt64(42)}, `{"foo":42}`},
+ {meta.Data{"foo": meta.String(`Hello "World"!`)}, `{"foo":"Hello \"World\"!"}`},
+ {meta.Data{"foo": meta.Entry{}}, `{"foo":null}`},
+ }
+
+ for _, tc := range cases {
+ got, err := json.Marshal(tc.d)
+ if err != nil {
+ t.Errorf("json.Marshal(%#v) = %v", tc.d, err)
+ continue
+ }
+
+ if diff := cmp.Diff(tc.want, string(got)); diff != "" {
+ t.Errorf("json.Marshal(%#v) differs (+got/-want):\n%s", tc.d, diff)
+ }
+ }
+}
+
+func TestUnmarshalJSON(t *testing.T) {
+ cases := []struct {
+ in string
+ want meta.Data
+ wantErr bool
+ }{
+ {
+ in: `{}`,
+ want: meta.Data{},
+ },
+ {
+ in: `{"bool":true}`,
+ want: meta.Data{"bool": meta.Bool(true)},
+ },
+ {
+ in: `{"string":"bar"}`,
+ want: meta.Data{"string": meta.String("bar")},
+ },
+ {
+ in: `{"int":42}`,
+ want: meta.Data{"int": meta.Int64(42)},
+ },
+ { // 9223372036854777144 exceeds 2^63-1
+ in: `{"uint":9223372036854777144}`,
+ want: meta.Data{"uint": meta.UInt64(9223372036854777144)},
+ },
+ {
+ in: `{"float":42.25}`,
+ want: meta.Data{"float": meta.Float64(42.25)},
+ },
+ {
+ in: `{"float":null}`,
+ want: meta.Data{"float": meta.Float64(math.NaN())},
+ },
+ {
+ in: `{"bool":false,"string":"","int":-9223372036854775808,"uint":18446744073709551615,"float":0.00006103515625}`,
+ want: meta.Data{
+ "bool": meta.Bool(false),
+ "string": meta.String(""),
+ "int": meta.Int64(-9223372036854775808),
+ "uint": meta.UInt64(18446744073709551615),
+ "float": meta.Float64(0.00006103515625),
+ },
+ },
+ {
+ in: `{"float":["invalid", "type"]}`,
+ wantErr: true,
+ },
+ }
+
+ for _, c := range cases {
+ var got meta.Data
+ err := json.Unmarshal([]byte(c.in), &got)
+ if gotErr := err != nil; gotErr != c.wantErr {
+ t.Errorf("Unmarshal() = %v, want error: %v", err, c.wantErr)
+ }
+ if err != nil || c.wantErr {
+ continue
+ }
+
+ opts := []cmp.Option{
+ cmp.AllowUnexported(meta.Entry{}),
+ cmpopts.EquateNaNs(),
+ }
+ if diff := cmp.Diff(c.want, got, opts...); diff != "" {
+ t.Errorf("Unmarshal() result differs (+got/-want):\n%s", diff)
+ }
+ }
+}
+
+func TestEntry(t *testing.T) {
+ cases := []struct {
+ typ string
+ e meta.Entry
+ wantBool bool
+ wantFloat64 bool
+ wantInt64 bool
+ wantUInt64 bool
+ wantString bool
+ s string
+ }{
+ {
+ typ: "bool",
+ e: meta.Bool(true),
+ wantBool: true,
+ s: "true",
+ },
+ {
+ typ: "float64",
+ e: meta.Float64(20.0 / 3.0),
+ wantFloat64: true,
+ s: "6.66666666666667",
+ },
+ {
+ typ: "int64",
+ e: meta.Int64(-9223372036854775808),
+ wantInt64: true,
+ s: "-9223372036854775808",
+ },
+ {
+ typ: "uint64",
+ e: meta.UInt64(18446744073709551615),
+ wantUInt64: true,
+ s: "18446744073709551615",
+ },
+ {
+ typ: "string",
+ e: meta.String("Hello, World!"),
+ wantString: true,
+ s: "Hello, World!",
+ },
+ {
+ // meta.Entry's zero value
+ typ: "<nil>",
+ s: "<nil>",
+ },
+ }
+
+ for _, tc := range cases {
+ if v, got := tc.e.Bool(); got != tc.wantBool {
+ t.Errorf("%#v.Bool() = (%v, %v), want (_, %v)", tc.e, v, got, tc.wantBool)
+ }
+
+ if v, got := tc.e.Float64(); got != tc.wantFloat64 {
+ t.Errorf("%#v.Float64() = (%v, %v), want (_, %v)", tc.e, v, got, tc.wantFloat64)
+ }
+
+ if v, got := tc.e.Int64(); got != tc.wantInt64 {
+ t.Errorf("%#v.Int64() = (%v, %v), want (_, %v)", tc.e, v, got, tc.wantInt64)
+ }
+
+ if v, got := tc.e.UInt64(); got != tc.wantUInt64 {
+ t.Errorf("%#v.UInt64() = (%v, %v), want (_, %v)", tc.e, v, got, tc.wantUInt64)
+ }
+
+ if got := tc.e.IsString(); got != tc.wantString {
+ t.Errorf("%#v.IsString() = %v, want %v", tc.e, got, tc.wantString)
+ }
+
+ if got, want := tc.e.String(), tc.s; got != want {
+ t.Errorf("%#v.String() = %q, want %q", tc.e, got, want)
+ }
+
+ if got, want := fmt.Sprintf("%T", tc.e.Interface()), tc.typ; got != want {
+ t.Errorf("%#v.Interface() = type %s, want type %s", tc.e, got, want)
+ }
+ }
+}
+
+func TestData_Clone(t *testing.T) {
+ want := meta.Data{
+ "bool": meta.Bool(false),
+ "string": meta.String(""),
+ "int": meta.Int64(-9223372036854775808),
+ "uint": meta.UInt64(18446744073709551615),
+ "float": meta.Float64(0.00006103515625),
+ }
+
+ got := want.Clone()
+
+ opts := []cmp.Option{
+ cmp.AllowUnexported(meta.Entry{}),
+ cmpopts.EquateNaNs(),
+ }
+ if diff := cmp.Diff(want, got, opts...); diff != "" {
+ t.Errorf("Data.Clone() contains differences (+got/-want):\n%s", diff)
+ }
+
+ want = nil
+ if got := meta.Data(nil).Clone(); got != nil {
+ t.Errorf("Data(nil).Clone() = %v, want %v", got, nil)
+ }
+}
diff --git a/network/buffer_test.go b/network/buffer_test.go
index cd838f0..aeaa655 100644
--- a/network/buffer_test.go
+++ b/network/buffer_test.go
@@ -3,6 +3,7 @@ package network // import "collectd.org/network"
import (
"bytes"
"context"
+ "errors"
"math"
"reflect"
"testing"
@@ -170,7 +171,7 @@ func TestUnknownType(t *testing.T) {
}
s1 := NewBuffer(0)
- if err := s1.Write(ctx, vl); err != ErrUnknownType {
+ if err := s1.Write(ctx, vl); !errors.Is(err, ErrUnknownType) {
t.Errorf("Buffer.Write(%v) = %v, want %v", vl, err, ErrUnknownType)
}
diff --git a/network/client.go b/network/client.go
index b3530c8..93df353 100644
--- a/network/client.go
+++ b/network/client.go
@@ -2,6 +2,7 @@ package network // import "collectd.org/network"
import (
"context"
+ "errors"
"net"
"collectd.org/api"
@@ -51,7 +52,7 @@ func Dial(address string, opts ClientOptions) (*Client, error) {
// Write adds a ValueList to the internal buffer. Data is only written to
// the network when the buffer is full.
func (c *Client) Write(ctx context.Context, vl *api.ValueList) error {
- if err := c.buffer.Write(ctx, vl); err != ErrNotEnoughSpace {
+ if err := c.buffer.Write(ctx, vl); !errors.Is(err, ErrNotEnoughSpace) {
return err
}
diff --git a/network/main.go b/network/main.go
index e08e841..7cc76f4 100644
--- a/network/main.go
+++ b/network/main.go
@@ -11,8 +11,8 @@ const (
DefaultPort = 25826
)
-// Default size of "Buffer". This is based on the maximum bytes that fit into
-// an Ethernet frame without fragmentation:
+// DefaultBufferSize is the default size of "Buffer". This is based on the
+// maximum bytes that fit into an Ethernet frame without fragmentation:
// <Ethernet frame> - (<IPv6 header> + <UDP header>) = 1500 - (40 + 8) = 1452
const DefaultBufferSize = 1452
diff --git a/network/network_x_test.go b/network/network_x_test.go
new file mode 100644
index 0000000..d6f4d41
--- /dev/null
+++ b/network/network_x_test.go
@@ -0,0 +1,119 @@
+package network_test
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "testing"
+ "time"
+
+ "collectd.org/api"
+ "collectd.org/network"
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+ "golang.org/x/net/nettest"
+)
+
+type testPasswordLookup map[string]string
+
+func (l testPasswordLookup) Password(user string) (string, error) {
+ pw, ok := l[user]
+ if !ok {
+ return "", fmt.Errorf("user %q not found", user)
+ }
+ return pw, nil
+}
+
+func TestNetwork(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ const (
+ username = "TestNetwork"
+ password = `oi5aGh7oLo0mai5oaG8zei8a`
+ )
+
+ conn, err := nettest.NewLocalPacketListener("udp")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+
+ ch := make(chan *api.ValueList)
+ go func() {
+ srv := &network.Server{
+ Conn: conn.(*net.UDPConn),
+ Writer: api.WriterFunc(func(_ context.Context, vl *api.ValueList) error {
+ ch <- vl
+ return nil
+ }),
+ PasswordLookup: testPasswordLookup{
+ username: password,
+ },
+ }
+
+ err := srv.ListenAndWrite(ctx)
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("Server.ListenAndWrite() = %v, want %v", err, context.Canceled)
+ }
+ close(ch)
+ }()
+
+ var want []*api.ValueList
+ go func() {
+ client, err := network.Dial(conn.LocalAddr().String(),
+ network.ClientOptions{
+ SecurityLevel: network.Encrypt,
+ Username: username,
+ Password: password,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ vl := &api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestNetwork",
+ Type: "gauge",
+ },
+ Time: time.Unix(1588164686, 0),
+ Interval: 10 * time.Second,
+ Values: []api.Value{api.Gauge(42)},
+ }
+
+ for i := 0; i < 30; i++ {
+ if err := client.Write(ctx, vl); err != nil {
+ t.Errorf("client.Write() = %v", err)
+ break
+ }
+ want = append(want, vl.Clone())
+
+ vl.Time = vl.Time.Add(vl.Interval)
+ }
+
+ if err := client.Close(); err != nil {
+ t.Errorf("client.Close() = %v", err)
+ }
+ }()
+
+ var got []*api.ValueList
+loop:
+ for {
+ select {
+ case vl, ok := <-ch:
+ if !ok {
+ break loop
+ }
+ got = append(got, vl)
+ case <-time.After(100 * time.Millisecond):
+ // cancel the context so the server returns.
+ cancel()
+ }
+ }
+
+ if diff := cmp.Diff(want, got, cmpopts.EquateEmpty()); diff != "" {
+ t.Errorf("sent and received value lists differ (+got/-want):\n%s", diff)
+ }
+}
diff --git a/network/parse.go b/network/parse.go
index 9027014..0779c58 100644
--- a/network/parse.go
+++ b/network/parse.go
@@ -110,7 +110,7 @@ func parse(b []byte, sl SecurityLevel, opts ParseOpts) ([]*api.ValueList, error)
// Returns an error if the number of values is incorrect.
v, err := ds.Values(ifValues...)
if err != nil {
- log.Printf("unable to convert values according to TypesDB: %v", err)
+ log.Printf("unable to convert metric %q, values %v according to %v in TypesDB: %v", state, ifValues, ds, err)
continue
}
vl.Values = v
diff --git a/network/server.go b/network/server.go
index 6197ea1..5bded61 100644
--- a/network/server.go
+++ b/network/server.go
@@ -4,6 +4,7 @@ import (
"context"
"log"
"net"
+ "sync"
"collectd.org/api"
)
@@ -44,13 +45,16 @@ type Server struct {
// Addr if Conn is nil), parses the received packets and writes them to the
// provided api.Writer.
func (srv *Server) ListenAndWrite(ctx context.Context) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
if srv.Conn == nil {
addr := srv.Addr
if addr == "" {
addr = ":" + DefaultService
}
- laddr, err := net.ResolveUDPAddr("udp", srv.Addr)
+ laddr, err := net.ResolveUDPAddr("udp", addr)
if err != nil {
return err
}
@@ -74,7 +78,6 @@ func (srv *Server) ListenAndWrite(ctx context.Context) error {
if srv.BufferSize <= 0 {
srv.BufferSize = DefaultBufferSize
}
- buf := make([]byte, srv.BufferSize)
popts := ParseOpts{
PasswordLookup: srv.PasswordLookup,
@@ -82,34 +85,24 @@ func (srv *Server) ListenAndWrite(ctx context.Context) error {
TypesDB: srv.TypesDB,
}
- var ctxErr error
- shutdown := make(chan struct{})
go func() {
select {
case <-ctx.Done():
- ctxErr = ctx.Err()
// this interrupts the below Conn.Read().
srv.Conn.Close()
- return
- case <-shutdown:
- return
}
}()
+ var wg sync.WaitGroup
for {
+ buf := make([]byte, srv.BufferSize)
n, err := srv.Conn.Read(buf)
if err != nil {
- // if ctxErr is non-nil the context got cancelled.
- if ctxErr != nil {
- srv.Conn = nil
- return ctxErr
- }
-
- // network error: shutdown the goroutine, close the
- // connection and return.
- close(shutdown)
srv.Conn.Close()
- srv.Conn = nil
+ wg.Wait()
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
return err
}
@@ -119,7 +112,11 @@ func (srv *Server) ListenAndWrite(ctx context.Context) error {
continue
}
- go dispatch(ctx, valueLists, srv.Writer)
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ dispatch(ctx, valueLists, srv.Writer)
+ }()
}
}
diff --git a/network/server_test.go b/network/server_test.go
index 05a6f24..caddcc6 100644
--- a/network/server_test.go
+++ b/network/server_test.go
@@ -2,6 +2,7 @@ package network // import "collectd.org/network"
import (
"context"
+ "errors"
"log"
"net"
"os"
@@ -52,7 +53,7 @@ func TestServer_Cancellation(t *testing.T) {
var srvErr error
go func() {
srv := &Server{
- Addr: "localhost:" + DefaultService,
+ Addr: "localhost:0",
}
srvErr = srv.ListenAndWrite(ctx)
@@ -64,7 +65,7 @@ func TestServer_Cancellation(t *testing.T) {
cancel()
wg.Wait()
- if srvErr != context.Canceled {
- t.Errorf("srvErr = %#v, want %#v", srvErr, context.Canceled)
+ if !errors.Is(srvErr, context.Canceled) {
+ t.Errorf("srvErr = %v, want %v", srvErr, context.Canceled)
}
}
diff --git a/plugin/README.md b/plugin/README.md
index 48ecd09..5d87860 100644
--- a/plugin/README.md
+++ b/plugin/README.md
@@ -2,8 +2,9 @@
## About
-This is _experimental_ code to write _collectd_ plugins in Go. It requires Go
-1.5 or later and a recent version of the collectd sources to build.
+This is _experimental_ code to write _collectd_ plugins in Go. That means the
+API is not yet stable. It requires Go 1.13 or later and a recent version of the
+collectd sources to build.
## Build
@@ -24,14 +25,13 @@ package.
## Future
-Only *read* and *write* callbacks are currently supported. Based on these
-implementations it should be fairly straightforward to implement the remaining
-callbacks. The *init*, *shutdown*, *log*, *flush* and *missing* callbacks are
-all likely low-hanging fruit. The *notification* callback is a bit trickier
-because it requires implementing notifications in the `collectd.org/api` package
-and the (un)marshaling of `notification_t`. The (complex) *config* callback is
-arguably the most important but, unfortunately, also the most complex to
-implemented.
+Only *log*, *read*, *write*, and *shutdown* callbacks are currently supported.
+Based on these implementations it should be possible to implement the remaining
+callbacks, even with little prior Cgo experience. The *init*, *flush*, and
+*missing* callbacks are all likely low-hanging fruit. The *notification*
+callback is a bit trickier because it requires implementing notifications in
+the `collectd.org/api` package and the (un)marshaling of `notification_t`. The
+(complex) *config* callback is currently work in progress, see #30.
If you're willing to give any of this a shot, please ping @octo to avoid
duplicate work.
diff --git a/plugin/c.go b/plugin/c.go
index feec9bf..07b42fd 100644
--- a/plugin/c.go
+++ b/plugin/c.go
@@ -8,35 +8,8 @@ package plugin // import "collectd.org/plugin"
// #include <stdlib.h>
// #include <dlfcn.h>
//
-// int (*register_read_ptr) (char const *group, char const *name,
-// plugin_read_cb callback,
-// cdtime_t interval,
-// user_data_t *ud) = NULL;
-// int register_read_wrapper (char const *group, char const *name,
-// plugin_read_cb callback,
-// cdtime_t interval,
-// user_data_t *ud) {
-// if (register_read_ptr == NULL) {
-// void *hnd = dlopen(NULL, RTLD_LAZY);
-// register_read_ptr = dlsym(hnd, "plugin_register_complex_read");
-// dlclose(hnd);
-// }
-// return (*register_read_ptr) (group, name, callback, interval, ud);
-// }
-//
-// int (*dispatch_values_ptr) (value_list_t const *vl);
-// int dispatch_values_wrapper (value_list_t const *vl) {
-// if (dispatch_values_ptr == NULL) {
-// void *hnd = dlopen(NULL, RTLD_LAZY);
-// dispatch_values_ptr = dlsym(hnd, "plugin_dispatch_values");
-// dlclose(hnd);
-// }
-// return (*dispatch_values_ptr) (vl);
-// }
-//
// void value_list_add (value_list_t *vl, value_t v) {
-// value_t *tmp;
-// tmp = realloc (vl->values, (vl->values_len + 1));
+// value_t *tmp = realloc (vl->values, sizeof(v) * (vl->values_len + 1));
// if (tmp == NULL) {
// errno = ENOMEM;
// return;
@@ -78,24 +51,35 @@ package plugin // import "collectd.org/plugin"
// return vl->values[i].derive;
// }
//
-// int (*register_write_ptr) (char const *, plugin_write_cb, user_data_t *);
-// int register_write_wrapper (char const *name, plugin_write_cb callback, user_data_t *user_data) {
-// if (register_write_ptr == NULL) {
+// static int *timeout_ptr;
+// int timeout_wrapper(void) {
+// if (timeout_ptr == NULL) {
// void *hnd = dlopen(NULL, RTLD_LAZY);
-// register_write_ptr = dlsym(hnd, "plugin_register_write");
+// timeout_ptr = dlsym(hnd, "timeout_g");
// dlclose(hnd);
// }
-// return (*register_write_ptr) (name, callback, user_data);
+// return *timeout_ptr;
// }
//
-// int (*register_shutdown_ptr) (char *, plugin_shutdown_cb);
-// int register_shutdown_wrapper (char *name, plugin_shutdown_cb callback) {
-// if (register_shutdown_ptr == NULL) {
+// typedef int (*plugin_complex_config_cb)(oconfig_item_t *);
+//
+// static int (*register_complex_config_ptr) (const char *, plugin_complex_config_cb);
+// int register_complex_config_wrapper (const char *name, plugin_complex_config_cb callback) {
+// if (register_complex_config_ptr == NULL) {
// void *hnd = dlopen(NULL, RTLD_LAZY);
-// register_shutdown_ptr = dlsym(hnd, "plugin_register_shutdown");
+// register_complex_config_ptr = dlsym(hnd, "plugin_register_complex_config");
// dlclose(hnd);
// }
-// return (*register_shutdown_ptr) (name, callback);
+// return (*register_complex_config_ptr) (name, callback);
+// }
//
+// static int (*register_init_ptr) (const char *, plugin_init_cb);
+// int register_init_wrapper (const char *name, plugin_init_cb callback) {
+// if (register_init_ptr == NULL) {
+// void *hnd = dlopen(NULL, RTLD_LAZY);
+// register_init_ptr = dlsym(hnd, "plugin_register_init");
+// dlclose(hnd);
+// }
+// return (*register_init_ptr) (name, callback);
// }
import "C"
diff --git a/plugin/config.go b/plugin/config.go
new file mode 100644
index 0000000..5a8c88c
--- /dev/null
+++ b/plugin/config.go
@@ -0,0 +1,125 @@
+package plugin
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include <stdbool.h>
+// #include <errno.h>
+// #include "plugin.h"
+//
+// /* work-around because Go can't deal with fields named "type". */
+// static int config_value_type(oconfig_value_t *v) {
+// if (v == NULL) {
+// errno = EINVAL;
+// return -1;
+// }
+// return v->type;
+// }
+//
+// /* work-around because CGo has trouble accessing unions. */
+// static char *config_value_string(oconfig_value_t *v) {
+// if (v == NULL || v->type != OCONFIG_TYPE_STRING) {
+// errno = EINVAL;
+// return NULL;
+// }
+// return v->value.string;
+// }
+// static double config_value_number(oconfig_value_t *v) {
+// if (v == NULL || v->type != OCONFIG_TYPE_NUMBER) {
+// errno = EINVAL;
+// return NAN;
+// }
+// return v->value.number;
+// }
+// static bool config_value_boolean(oconfig_value_t *v) {
+// if (v == NULL || v->type != OCONFIG_TYPE_BOOLEAN) {
+// errno = EINVAL;
+// return 0;
+// }
+// return v->value.boolean;
+// }
+import "C"
+
+import (
+ "fmt"
+ "unsafe"
+
+ "collectd.org/config"
+)
+
+func unmarshalConfigBlocks(blocks *C.oconfig_item_t, blocksNum C.int) ([]config.Block, error) {
+ var ret []config.Block
+ for i := C.int(0); i < blocksNum; i++ {
+ offset := uintptr(i) * C.sizeof_oconfig_item_t
+ cBlock := (*C.oconfig_item_t)(unsafe.Pointer(uintptr(unsafe.Pointer(blocks)) + offset))
+
+ goBlock, err := unmarshalConfigBlock(cBlock)
+ if err != nil {
+ return nil, err
+ }
+ ret = append(ret, goBlock)
+ }
+ return ret, nil
+}
+
+func unmarshalConfigBlock(block *C.oconfig_item_t) (config.Block, error) {
+ cfg := config.Block{
+ Key: C.GoString(block.key),
+ }
+
+ var err error
+ if cfg.Values, err = unmarshalConfigValues(block.values, block.values_num); err != nil {
+ return config.Block{}, err
+ }
+
+ if cfg.Children, err = unmarshalConfigBlocks(block.children, block.children_num); err != nil {
+ return config.Block{}, err
+ }
+
+ return cfg, nil
+}
+
+func unmarshalConfigValues(values *C.oconfig_value_t, valuesNum C.int) ([]config.Value, error) {
+ var ret []config.Value
+ for i := C.int(0); i < valuesNum; i++ {
+ offset := uintptr(i) * C.sizeof_oconfig_value_t
+ cValue := (*C.oconfig_value_t)(unsafe.Pointer(uintptr(unsafe.Pointer(values)) + offset))
+
+ goValue, err := unmarshalConfigValue(cValue)
+ if err != nil {
+ return nil, err
+ }
+ ret = append(ret, goValue)
+ }
+ return ret, nil
+}
+
+func unmarshalConfigValue(value *C.oconfig_value_t) (config.Value, error) {
+ typ, err := C.config_value_type(value)
+ if err := wrapCError(0, err, "config_value_type"); err != nil {
+ return config.Value{}, err
+ }
+
+ switch typ {
+ case C.OCONFIG_TYPE_STRING:
+ s, err := C.config_value_string(value)
+ if err := wrapCError(0, err, "config_value_string"); err != nil {
+ return config.Value{}, err
+ }
+ return config.String(C.GoString(s)), nil
+ case C.OCONFIG_TYPE_NUMBER:
+ n, err := C.config_value_number(value)
+ if err := wrapCError(0, err, "config_value_number"); err != nil {
+ return config.Value{}, err
+ }
+ return config.Float64(float64(n)), nil
+ case C.OCONFIG_TYPE_BOOLEAN:
+ b, err := C.config_value_boolean(value)
+ if err := wrapCError(0, err, "config_value_boolean"); err != nil {
+ return config.Value{}, err
+ }
+ return config.Bool(bool(b)), nil
+ default:
+ return config.Value{}, fmt.Errorf("unknown config value type: %d", typ)
+ }
+}
diff --git a/plugin/fake/fake.go b/plugin/fake/fake.go
new file mode 100644
index 0000000..d7a53a7
--- /dev/null
+++ b/plugin/fake/fake.go
@@ -0,0 +1,29 @@
+// Package fake implements fake versions of the C functions imported from the
+// collectd daemon for testing.
+package fake
+
+// void reset_log(void);
+// void reset_read(void);
+// void reset_shutdown(void);
+// void reset_write(void);
+//
+// int timeout_g = 2;
+import "C"
+
+import (
+ "time"
+)
+
+// TearDown cleans up after a test and prepares shared resources for the next
+// test.
+//
+// Note that this only resets the state of the fake implementations, such as
+// "plugin_register_log()". The Go code in "collectd.org/plugin" may still hold
+// a reference to the callback even after this function has been called.
+func TearDown() {
+ SetInterval(10 * time.Second)
+ C.reset_log()
+ C.reset_read()
+ C.reset_shutdown()
+ C.reset_write()
+}
diff --git a/plugin/fake/interval.go b/plugin/fake/interval.go
new file mode 100644
index 0000000..ac881c8
--- /dev/null
+++ b/plugin/fake/interval.go
@@ -0,0 +1,28 @@
+package fake
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include "plugin.h"
+//
+// static cdtime_t interval = TIME_T_TO_CDTIME_T_STATIC(10);
+// cdtime_t plugin_get_interval(void) {
+// return interval;
+// }
+// void plugin_set_interval(cdtime_t d) {
+// interval = d;
+// }
+import "C"
+
+import (
+ "time"
+
+ "collectd.org/cdtime"
+)
+
+// SetInterval sets the interval returned by the fake plugin_get_interval()
+// function.
+func SetInterval(d time.Duration) {
+ ival := cdtime.NewDuration(d)
+ C.plugin_set_interval(C.cdtime_t(ival))
+}
diff --git a/plugin/fake/log.go b/plugin/fake/log.go
new file mode 100644
index 0000000..ed3e1e2
--- /dev/null
+++ b/plugin/fake/log.go
@@ -0,0 +1,60 @@
+package fake
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include "plugin.h"
+//
+// typedef struct {
+// const char *name;
+// plugin_log_cb callback;
+// user_data_t user_data;
+// } log_callback_t;
+// static log_callback_t *log_callbacks = NULL;
+// static size_t log_callbacks_num = 0;
+//
+// int plugin_register_log(const char *name,
+// plugin_log_cb callback,
+// user_data_t const *user_data) {
+// log_callback_t *ptr = realloc(log_callbacks, (log_callbacks_num+1) * sizeof(*log_callbacks));
+// if (ptr == NULL) {
+// return ENOMEM;
+// }
+// log_callbacks = ptr;
+// log_callbacks[log_callbacks_num] = (log_callback_t){
+// .name = name,
+// .callback = callback,
+// .user_data = *user_data,
+// };
+// log_callbacks_num++;
+//
+// return 0;
+// }
+//
+// void plugin_log(int level, const char *format, ...) {
+// char msg[1024];
+// va_list ap;
+// va_start(ap, format);
+// vsnprintf(msg, sizeof(msg), format, ap);
+// msg[sizeof(msg)-1] = 0;
+// va_end(ap);
+//
+// for (size_t i = 0; i < log_callbacks_num; i++) {
+// log_callbacks[i].callback(level, msg, &log_callbacks[i].user_data);
+// }
+// }
+//
+// void reset_log(void) {
+// for (size_t i = 0; i < log_callbacks_num; i++) {
+// user_data_t *ud = &log_callbacks[i].user_data;
+// if (ud->free_func == NULL) {
+// continue;
+// }
+// ud->free_func(ud->data);
+// ud->data = NULL;
+// }
+// free(log_callbacks);
+// log_callbacks = NULL;
+// log_callbacks_num = 0;
+// }
+import "C"
diff --git a/plugin/fake/meta_data.c b/plugin/fake/meta_data.c
new file mode 100644
index 0000000..393ce0d
--- /dev/null
+++ b/plugin/fake/meta_data.c
@@ -0,0 +1,487 @@
+/**
+ * collectd - src/meta_data.c
+ * Copyright (C) 2008-2011 Florian octo Forster
+ *
+ * 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.
+ *
+ * Authors:
+ * Florian octo Forster <octo at collectd.org>
+ **/
+
+#include "collectd.h"
+#include "plugin.h"
+
+#include <stdbool.h>
+
+#define MD_MAX_NONSTRING_CHARS 128
+
+/*
+ * Data types
+ */
+union meta_value_u {
+ char *mv_string;
+ int64_t mv_signed_int;
+ uint64_t mv_unsigned_int;
+ double mv_double;
+ bool mv_boolean;
+};
+typedef union meta_value_u meta_value_t;
+
+struct meta_entry_s;
+typedef struct meta_entry_s meta_entry_t;
+struct meta_entry_s {
+ char *key;
+ meta_value_t value;
+ int type;
+ meta_entry_t *next;
+};
+
+struct meta_data_s {
+ meta_entry_t *head;
+ pthread_mutex_t lock;
+};
+
+/*
+ * Private functions
+ */
+static meta_entry_t *md_entry_alloc(const char *key) /* {{{ */
+{
+ meta_entry_t *e;
+
+ e = calloc(1, sizeof(*e));
+ if (e == NULL) {
+ ERROR("md_entry_alloc: calloc failed.");
+ return NULL;
+ }
+
+ e->key = strdup(key);
+ if (e->key == NULL) {
+ free(e);
+ ERROR("md_entry_alloc: strdup failed.");
+ return NULL;
+ }
+
+ e->type = 0;
+ e->next = NULL;
+
+ return e;
+} /* }}} meta_entry_t *md_entry_alloc */
+
+static void md_entry_free(meta_entry_t *e) /* {{{ */
+{
+ if (e == NULL)
+ return;
+
+ free(e->key);
+
+ if (e->type == MD_TYPE_STRING)
+ free(e->value.mv_string);
+
+ if (e->next != NULL)
+ md_entry_free(e->next);
+
+ free(e);
+} /* }}} void md_entry_free */
+
+static int md_entry_insert(meta_data_t *md, meta_entry_t *e) /* {{{ */
+{
+ meta_entry_t *this;
+ meta_entry_t *prev;
+
+ if ((md == NULL) || (e == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ prev = NULL;
+ this = md->head;
+ while (this != NULL) {
+ if (strcasecmp(e->key, this->key) == 0)
+ break;
+
+ prev = this;
+ this = this->next;
+ }
+
+ if (this == NULL) {
+ /* This key does not exist yet. */
+ if (md->head == NULL)
+ md->head = e;
+ else {
+ assert(prev != NULL);
+ prev->next = e;
+ }
+
+ e->next = NULL;
+ } else /* (this != NULL) */
+ {
+ if (prev == NULL)
+ md->head = e;
+ else
+ prev->next = e;
+
+ e->next = this->next;
+ }
+
+ pthread_mutex_unlock(&md->lock);
+
+ if (this != NULL) {
+ this->next = NULL;
+ md_entry_free(this);
+ }
+
+ return 0;
+} /* }}} int md_entry_insert */
+
+/* XXX: The lock on md must be held while calling this function! */
+static meta_entry_t *md_entry_lookup(meta_data_t *md, /* {{{ */
+ const char *key) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL))
+ return NULL;
+
+ for (e = md->head; e != NULL; e = e->next)
+ if (strcasecmp(key, e->key) == 0)
+ break;
+
+ return e;
+} /* }}} meta_entry_t *md_entry_lookup */
+
+/*
+ * Each value_list_t*, as it is going through the system, is handled by exactly
+ * one thread. Plugins which pass a value_list_t* to another thread, e.g. the
+ * rrdtool plugin, must create a copy first. The meta data within a
+ * value_list_t* is not thread safe and doesn't need to be.
+ *
+ * The meta data associated with cache entries are a different story. There, we
+ * need to ensure exclusive locking to prevent leaks and other funky business.
+ * This is ensured by the uc_meta_data_get_*() functions.
+ */
+
+/*
+ * Public functions
+ */
+meta_data_t *meta_data_create(void) /* {{{ */
+{
+ meta_data_t *md;
+
+ md = calloc(1, sizeof(*md));
+ if (md == NULL) {
+ ERROR("meta_data_create: calloc failed.");
+ return NULL;
+ }
+
+ pthread_mutex_init(&md->lock, /* attr = */ NULL);
+
+ return md;
+} /* }}} meta_data_t *meta_data_create */
+
+void meta_data_destroy(meta_data_t *md) /* {{{ */
+{
+ if (md == NULL)
+ return;
+
+ md_entry_free(md->head);
+ pthread_mutex_destroy(&md->lock);
+ free(md);
+} /* }}} void meta_data_destroy */
+
+int meta_data_type(meta_data_t *md, const char *key) /* {{{ */
+{
+ if ((md == NULL) || (key == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ for (meta_entry_t *e = md->head; e != NULL; e = e->next) {
+ if (strcasecmp(key, e->key) == 0) {
+ pthread_mutex_unlock(&md->lock);
+ return e->type;
+ }
+ }
+
+ pthread_mutex_unlock(&md->lock);
+ return 0;
+} /* }}} int meta_data_type */
+
+int meta_data_toc(meta_data_t *md, char ***toc) /* {{{ */
+{
+ int i = 0, count = 0;
+
+ if ((md == NULL) || (toc == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ for (meta_entry_t *e = md->head; e != NULL; e = e->next)
+ ++count;
+
+ if (count == 0) {
+ pthread_mutex_unlock(&md->lock);
+ return count;
+ }
+
+ *toc = calloc(count, sizeof(**toc));
+ for (meta_entry_t *e = md->head; e != NULL; e = e->next)
+ (*toc)[i++] = strdup(e->key);
+
+ pthread_mutex_unlock(&md->lock);
+ return count;
+} /* }}} int meta_data_toc */
+
+/*
+ * Add functions
+ */
+int meta_data_add_string(meta_data_t *md, /* {{{ */
+ const char *key, const char *value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ e = md_entry_alloc(key);
+ if (e == NULL)
+ return -ENOMEM;
+
+ e->value.mv_string = strdup(value);
+ if (e->value.mv_string == NULL) {
+ ERROR("meta_data_add_string: strdup failed.");
+ md_entry_free(e);
+ return -ENOMEM;
+ }
+ e->type = MD_TYPE_STRING;
+
+ return md_entry_insert(md, e);
+} /* }}} int meta_data_add_string */
+
+int meta_data_add_signed_int(meta_data_t *md, /* {{{ */
+ const char *key, int64_t value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL))
+ return -EINVAL;
+
+ e = md_entry_alloc(key);
+ if (e == NULL)
+ return -ENOMEM;
+
+ e->value.mv_signed_int = value;
+ e->type = MD_TYPE_SIGNED_INT;
+
+ return md_entry_insert(md, e);
+} /* }}} int meta_data_add_signed_int */
+
+int meta_data_add_unsigned_int(meta_data_t *md, /* {{{ */
+ const char *key, uint64_t value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL))
+ return -EINVAL;
+
+ e = md_entry_alloc(key);
+ if (e == NULL)
+ return -ENOMEM;
+
+ e->value.mv_unsigned_int = value;
+ e->type = MD_TYPE_UNSIGNED_INT;
+
+ return md_entry_insert(md, e);
+} /* }}} int meta_data_add_unsigned_int */
+
+int meta_data_add_double(meta_data_t *md, /* {{{ */
+ const char *key, double value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL))
+ return -EINVAL;
+
+ e = md_entry_alloc(key);
+ if (e == NULL)
+ return -ENOMEM;
+
+ e->value.mv_double = value;
+ e->type = MD_TYPE_DOUBLE;
+
+ return md_entry_insert(md, e);
+} /* }}} int meta_data_add_double */
+
+int meta_data_add_boolean(meta_data_t *md, /* {{{ */
+ const char *key, bool value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL))
+ return -EINVAL;
+
+ e = md_entry_alloc(key);
+ if (e == NULL)
+ return -ENOMEM;
+
+ e->value.mv_boolean = value;
+ e->type = MD_TYPE_BOOLEAN;
+
+ return md_entry_insert(md, e);
+} /* }}} int meta_data_add_boolean */
+
+/*
+ * Get functions
+ */
+int meta_data_get_string(meta_data_t *md, /* {{{ */
+ const char *key, char **value) {
+ meta_entry_t *e;
+ char *temp;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ e = md_entry_lookup(md, key);
+ if (e == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ if (e->type != MD_TYPE_STRING) {
+ ERROR("meta_data_get_string: Type mismatch for key `%s'", e->key);
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ temp = strdup(e->value.mv_string);
+ if (temp == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ ERROR("meta_data_get_string: strdup failed.");
+ return -ENOMEM;
+ }
+
+ pthread_mutex_unlock(&md->lock);
+
+ *value = temp;
+
+ return 0;
+} /* }}} int meta_data_get_string */
+
+int meta_data_get_signed_int(meta_data_t *md, /* {{{ */
+ const char *key, int64_t *value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ e = md_entry_lookup(md, key);
+ if (e == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ if (e->type != MD_TYPE_SIGNED_INT) {
+ ERROR("meta_data_get_signed_int: Type mismatch for key `%s'", e->key);
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ *value = e->value.mv_signed_int;
+
+ pthread_mutex_unlock(&md->lock);
+ return 0;
+} /* }}} int meta_data_get_signed_int */
+
+int meta_data_get_unsigned_int(meta_data_t *md, /* {{{ */
+ const char *key, uint64_t *value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ e = md_entry_lookup(md, key);
+ if (e == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ if (e->type != MD_TYPE_UNSIGNED_INT) {
+ ERROR("meta_data_get_unsigned_int: Type mismatch for key `%s'", e->key);
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ *value = e->value.mv_unsigned_int;
+
+ pthread_mutex_unlock(&md->lock);
+ return 0;
+} /* }}} int meta_data_get_unsigned_int */
+
+int meta_data_get_double(meta_data_t *md, /* {{{ */
+ const char *key, double *value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ e = md_entry_lookup(md, key);
+ if (e == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ if (e->type != MD_TYPE_DOUBLE) {
+ ERROR("meta_data_get_double: Type mismatch for key `%s'", e->key);
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ *value = e->value.mv_double;
+
+ pthread_mutex_unlock(&md->lock);
+ return 0;
+} /* }}} int meta_data_get_double */
+
+int meta_data_get_boolean(meta_data_t *md, /* {{{ */
+ const char *key, bool *value) {
+ meta_entry_t *e;
+
+ if ((md == NULL) || (key == NULL) || (value == NULL))
+ return -EINVAL;
+
+ pthread_mutex_lock(&md->lock);
+
+ e = md_entry_lookup(md, key);
+ if (e == NULL) {
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ if (e->type != MD_TYPE_BOOLEAN) {
+ ERROR("meta_data_get_boolean: Type mismatch for key `%s'", e->key);
+ pthread_mutex_unlock(&md->lock);
+ return -ENOENT;
+ }
+
+ *value = e->value.mv_boolean;
+
+ pthread_mutex_unlock(&md->lock);
+ return 0;
+} /* }}} int meta_data_get_boolean */
diff --git a/plugin/fake/read.go b/plugin/fake/read.go
new file mode 100644
index 0000000..03fafb5
--- /dev/null
+++ b/plugin/fake/read.go
@@ -0,0 +1,120 @@
+package fake
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include <string.h>
+// #include "plugin.h"
+//
+// typedef struct {
+// char *group;
+// char *name;
+// plugin_read_cb callback;
+// cdtime_t interval;
+// user_data_t user_data;
+// } read_callback_t;
+// read_callback_t *read_callbacks = NULL;
+// size_t read_callbacks_num = 0;
+//
+// int plugin_register_complex_read(const char *group, const char *name,
+// plugin_read_cb callback, cdtime_t interval,
+// user_data_t const *user_data) {
+// if (interval == 0) {
+// interval = plugin_get_interval();
+// }
+//
+// read_callback_t *ptr = realloc(
+// read_callbacks, (read_callbacks_num + 1) * sizeof(*read_callbacks));
+// if (ptr == NULL) {
+// return ENOMEM;
+// }
+// read_callbacks = ptr;
+// read_callbacks[read_callbacks_num] = (read_callback_t){
+// .group = (group != NULL) ? strdup(group) : NULL,
+// .name = strdup(name),
+// .callback = callback,
+// .interval = interval,
+// .user_data = *user_data,
+// };
+// read_callbacks_num++;
+//
+// return 0;
+// }
+//
+// void plugin_set_interval(cdtime_t);
+// static int read_all(void) {
+// cdtime_t save_interval = plugin_get_interval();
+// int ret = 0;
+//
+// for (size_t i = 0; i < read_callbacks_num; i++) {
+// read_callback_t *cb = read_callbacks + i;
+// plugin_set_interval(cb->interval);
+// int err = cb->callback(&cb->user_data);
+// if (err != 0) {
+// ret = err;
+// }
+// }
+//
+// plugin_set_interval(save_interval);
+// return ret;
+// }
+//
+// void reset_read(void) {
+// for (size_t i = 0; i < read_callbacks_num; i++) {
+// free(read_callbacks[i].name);
+// free(read_callbacks[i].group);
+// user_data_t *ud = &read_callbacks[i].user_data;
+// if (ud->free_func == NULL) {
+// continue;
+// }
+// ud->free_func(ud->data);
+// ud->data = NULL;
+// }
+// free(read_callbacks);
+// read_callbacks = NULL;
+// read_callbacks_num = 0;
+// }
+import "C"
+
+import (
+ "fmt"
+ "unsafe"
+
+ "collectd.org/cdtime"
+)
+
+func ReadAll() error {
+ status, err := C.read_all()
+ if err != nil {
+ return err
+ }
+ if status != 0 {
+ return fmt.Errorf("read_all() = %d", status)
+ }
+
+ return nil
+}
+
+// ReadCallback represents a data associated with a registered read callback.
+type ReadCallback struct {
+ Group, Name string
+ Interval cdtime.Time
+}
+
+// ReadCallbacks returns the data associated with all registered read
+// callbacks.
+func ReadCallbacks() []ReadCallback {
+ var ret []ReadCallback
+
+ for i := C.size_t(0); i < C.read_callbacks_num; i++ {
+ // Go pointer arithmetic that does the equivalent of C's `read_callbacks[i]`.
+ cb := (*C.read_callback_t)(unsafe.Pointer(uintptr(unsafe.Pointer(C.read_callbacks)) + uintptr(C.sizeof_read_callback_t*i)))
+ ret = append(ret, ReadCallback{
+ Group: C.GoString(cb.group),
+ Name: C.GoString(cb.name),
+ Interval: cdtime.Time(cb.interval),
+ })
+ }
+
+ return ret
+}
diff --git a/plugin/fake/shutdown.go b/plugin/fake/shutdown.go
new file mode 100644
index 0000000..02432e5
--- /dev/null
+++ b/plugin/fake/shutdown.go
@@ -0,0 +1,65 @@
+package fake
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include "plugin.h"
+//
+// typedef struct {
+// const char *name;
+// plugin_shutdown_cb callback;
+// } shutdown_callback_t;
+// static shutdown_callback_t *shutdown_callbacks = NULL;
+// static size_t shutdown_callbacks_num = 0;
+//
+// int plugin_register_shutdown(const char *name, plugin_shutdown_cb callback) {
+// shutdown_callback_t *ptr =
+// realloc(shutdown_callbacks,
+// (shutdown_callbacks_num + 1) * sizeof(*shutdown_callbacks));
+// if (ptr == NULL) {
+// return ENOMEM;
+// }
+// shutdown_callbacks = ptr;
+// shutdown_callbacks[shutdown_callbacks_num] = (shutdown_callback_t){
+// .name = name,
+// .callback = callback,
+// };
+// shutdown_callbacks_num++;
+//
+// return 0;
+// }
+//
+// int plugin_shutdown_all(void) {
+// int ret = 0;
+// for (size_t i = 0; i < shutdown_callbacks_num; i++) {
+// int err = shutdown_callbacks[i].callback();
+// if (err != 0) {
+// ret = err;
+// }
+// }
+// return ret;
+// }
+//
+// void reset_shutdown(void) {
+// free(shutdown_callbacks);
+// shutdown_callbacks = NULL;
+// shutdown_callbacks_num = 0;
+// }
+import "C"
+
+import (
+ "fmt"
+)
+
+// ShutdownAll calls all registered shutdown callbacks.
+func ShutdownAll() error {
+ status, err := C.plugin_shutdown_all()
+ if err != nil {
+ return err
+ }
+ if status != 0 {
+ return fmt.Errorf("plugin_shutdown_all() = %d", status)
+ }
+
+ return nil
+}
diff --git a/plugin/fake/write.go b/plugin/fake/write.go
new file mode 100644
index 0000000..18db388
--- /dev/null
+++ b/plugin/fake/write.go
@@ -0,0 +1,85 @@
+package fake
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include <stdlib.h>
+// #include <stdio.h>
+// #include "plugin.h"
+//
+// typedef struct {
+// const char *name;
+// plugin_write_cb callback;
+// user_data_t user_data;
+// } write_callback_t;
+// static write_callback_t *write_callbacks = NULL;
+// static size_t write_callbacks_num = 0;
+//
+// int plugin_register_write(const char *name, plugin_write_cb callback,
+// user_data_t const *user_data) {
+// write_callback_t *ptr = realloc(
+// write_callbacks, (write_callbacks_num + 1) * sizeof(*write_callbacks));
+// if (ptr == NULL) {
+// return ENOMEM;
+// }
+// write_callbacks = ptr;
+// write_callbacks[write_callbacks_num] = (write_callback_t){
+// .name = name,
+// .callback = callback,
+// .user_data = *user_data,
+// };
+// write_callbacks_num++;
+//
+// return 0;
+// }
+//
+// int plugin_dispatch_values(value_list_t const *vl) {
+// data_set_t *ds = &(data_set_t){
+// .ds_num = 1,
+// .ds =
+// &(data_source_t){
+// .name = "value",
+// .min = 0,
+// .max = NAN,
+// },
+// };
+//
+// if (strcmp("derive", vl->type) == 0) {
+// strncpy(ds->type, vl->type, sizeof(ds->type));
+// ds->ds[0].type = DS_TYPE_DERIVE;
+// } else if (strcmp("gauge", vl->type) == 0) {
+// strncpy(ds->type, vl->type, sizeof(ds->type));
+// ds->ds[0].type = DS_TYPE_GAUGE;
+// } else if (strcmp("counter", vl->type) == 0) {
+// strncpy(ds->type, vl->type, sizeof(ds->type));
+// ds->ds[0].type = DS_TYPE_COUNTER;
+// } else {
+// errno = EINVAL;
+// return errno;
+// }
+//
+// int ret = 0;
+// for (size_t i = 0; i < write_callbacks_num; i++) {
+// int err =
+// write_callbacks[i].callback(ds, vl, &write_callbacks[i].user_data);
+// if (err != 0) {
+// ret = err;
+// }
+// }
+//
+// return ret;
+// }
+//
+// void reset_write(void) {
+// for (size_t i = 0; i < write_callbacks_num; i++) {
+// user_data_t *ud = &write_callbacks[i].user_data;
+// if (ud->free_func == NULL) {
+// continue;
+// }
+// ud->free_func(ud->data);
+// ud->data = NULL;
+// }
+// free(write_callbacks);
+// write_callbacks = NULL;
+// write_callbacks_num = 0;
+// }
+import "C"
diff --git a/plugin/generator/generator.go b/plugin/generator/generator.go
new file mode 100644
index 0000000..28417f9
--- /dev/null
+++ b/plugin/generator/generator.go
@@ -0,0 +1,314 @@
+package main
+
+import (
+ "bufio"
+ "bytes"
+ "fmt"
+ "log"
+ "os"
+ "os/exec"
+ "sort"
+ "strings"
+ "text/template"
+)
+
+var functions = []Function{
+ {
+ Name: "plugin_register_complex_read",
+ Args: []Argument{
+ {"group", "meta_data_t *"},
+ {"name", "char const *"},
+ {"callback", "plugin_read_cb"},
+ {"interval", "cdtime_t"},
+ {"ud", "user_data_t *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "plugin_register_write",
+ Args: []Argument{
+ {"name", "char const *"},
+ {"callback", "plugin_write_cb"},
+ {"ud", "user_data_t *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "plugin_register_shutdown",
+ Args: []Argument{
+ {"name", "char const *"},
+ {"callback", "plugin_shutdown_cb"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "plugin_register_log",
+ Args: []Argument{
+ {"name", "char const *"},
+ {"callback", "plugin_log_cb"},
+ {"ud", "user_data_t *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "plugin_dispatch_values",
+ Args: []Argument{
+ {"vl", "value_list_t const *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "plugin_get_interval",
+ Ret: "cdtime_t",
+ },
+ {
+ Name: "meta_data_create",
+ Ret: "meta_data_t *",
+ },
+ {
+ Name: "meta_data_destroy",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ },
+ Ret: "void",
+ },
+ {
+ Name: "meta_data_toc",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"toc", "char ***"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_type",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_add_boolean",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "bool"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_add_double",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "double"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_add_signed_int",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "int64_t"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_add_unsigned_int",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "uint64_t"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_add_string",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "char const *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_get_boolean",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "bool *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_get_double",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "double *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_get_signed_int",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "int64_t *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_get_unsigned_int",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "uint64_t *"},
+ },
+ Ret: "int",
+ },
+ {
+ Name: "meta_data_get_string",
+ Args: []Argument{
+ {"md", "meta_data_t *"},
+ {"key", "char const *"},
+ {"value", "char **"},
+ },
+ Ret: "int",
+ },
+}
+
+const ptrTmpl = "static {{.Ret}} (*{{.Name}}_ptr)({{.ArgsTypes}});\n"
+
+const wrapperTmpl = `{{.Ret}} {{.Name}}_wrapper({{.ArgsStr}}) {
+ LOAD({{.Name}});
+ {{if not .IsVoid}}return {{end}}(*{{.Name}}_ptr)({{.ArgsNames}});
+}
+
+`
+
+type Argument struct {
+ Name string
+ Type string
+}
+
+func (a Argument) String() string {
+ return a.Type + " " + a.Name
+}
+
+type Function struct {
+ Name string
+ Args []Argument
+ Ret string
+}
+
+func (f Function) ArgsTypes() string {
+ if len(f.Args) == 0 {
+ return "void"
+ }
+
+ var args []string
+ for _, a := range f.Args {
+ args = append(args, a.Type)
+ }
+
+ return strings.Join(args, ", ")
+}
+
+func (f Function) ArgsNames() string {
+ var args []string
+ for _, a := range f.Args {
+ args = append(args, a.Name)
+ }
+
+ return strings.Join(args, ", ")
+}
+
+func (f Function) ArgsStr() string {
+ if len(f.Args) == 0 {
+ return "void"
+ }
+
+ var args []string
+ for _, a := range f.Args {
+ args = append(args, a.String())
+ }
+
+ return strings.Join(args, ", ")
+}
+
+func (f Function) IsVoid() bool {
+ return f.Ret == "void"
+}
+
+type byName []Function
+
+func (f byName) Len() int { return len(f) }
+func (f byName) Less(i, j int) bool { return f[i].Name < f[j].Name }
+func (f byName) Swap(i, j int) { f[i], f[j] = f[j], f[i] }
+
+func main() {
+ var rawC bytes.Buffer
+ fmt.Fprint(&rawC, `#include "plugin.h"
+#include <stdlib.h>
+#include <stdbool.h>
+#include <dlfcn.h>
+
+#define LOAD(f) \
+ if (f##_ptr == NULL) { \
+ void *hnd = dlopen(NULL, RTLD_LAZY); \
+ f##_ptr = dlsym(hnd, #f); \
+ dlclose(hnd); \
+ }
+
+`)
+
+ sort.Sort(byName(functions))
+
+ t, err := template.New("ptr").Parse(ptrTmpl)
+ if err != nil {
+ log.Fatal(err)
+ }
+ for _, f := range functions {
+ if err := t.Execute(&rawC, f); err != nil {
+ log.Fatal(err)
+ }
+ }
+
+ fmt.Fprintln(&rawC)
+
+ t, err = template.New("wrapper").Parse(wrapperTmpl)
+ if err != nil {
+ log.Fatal(err)
+ }
+ for _, f := range functions {
+ if err := t.Execute(&rawC, f); err != nil {
+ log.Fatal(err)
+ }
+ }
+
+ var fmtC bytes.Buffer
+
+ cmd := exec.Command("clang-format")
+ cmd.Stdin = &rawC
+ cmd.Stdout = &fmtC
+ cmd.Stderr = os.Stderr
+ if err := cmd.Run(); err != nil {
+ log.Fatal(err)
+ }
+
+ fmt.Print(`// +build go1.5,cgo
+
+package plugin // import "collectd.org/plugin"
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+`)
+ s := bufio.NewScanner(&fmtC)
+ for s.Scan() {
+ fmt.Println("//", s.Text())
+ }
+ fmt.Println(`import "C"`)
+}
diff --git a/plugin/log.go b/plugin/log.go
index a7c7a5c..a4767dd 100644
--- a/plugin/log.go
+++ b/plugin/log.go
@@ -21,31 +21,40 @@ import "C"
import (
"fmt"
+ "strings"
+ "unicode"
"unsafe"
)
-type severity int
+// Severity is the severity of log messages. These are well-known constants
+// within collectd, so don't define your own. Use the constants provided by
+// this package instead.
+type Severity int
+// Predefined severities for collectd log functions.
const (
- logErr severity = 3
- logWarning severity = 4
- logNotice severity = 5
- logInfo severity = 6
- logDebug severity = 7
+ SeverityError Severity = 3
+ SeverityWarning Severity = 4
+ SeverityNotice Severity = 5
+ SeverityInfo Severity = 6
+ SeverityDebug Severity = 7
)
-func log(s severity, msg string) error {
+func log(s Severity, msg string) error {
+ // Trim trailing whitespace.
+ msg = strings.TrimRightFunc(msg, unicode.IsSpace)
+
ptr := C.CString(msg)
defer C.free(unsafe.Pointer(ptr))
_, err := C.wrap_plugin_log(C.int(s), ptr)
- return err
+ return wrapCError(0, err, "plugin_log")
}
// Error logs an error using plugin_log(). Arguments are handled in the manner
// of fmt.Print.
func Error(v ...interface{}) error {
- return log(logErr, fmt.Sprint(v...))
+ return log(SeverityError, fmt.Sprint(v...))
}
// Errorf logs an error using plugin_log(). Arguments are handled in the manner
@@ -57,7 +66,7 @@ func Errorf(format string, v ...interface{}) error {
// Warning logs a warning using plugin_log(). Arguments are handled in the
// manner of fmt.Print.
func Warning(v ...interface{}) error {
- return log(logWarning, fmt.Sprint(v...))
+ return log(SeverityWarning, fmt.Sprint(v...))
}
// Warningf logs a warning using plugin_log(). Arguments are handled in the
@@ -69,7 +78,7 @@ func Warningf(format string, v ...interface{}) error {
// Notice logs a notice using plugin_log(). Arguments are handled in the manner
// of fmt.Print.
func Notice(v ...interface{}) error {
- return log(logNotice, fmt.Sprint(v...))
+ return log(SeverityNotice, fmt.Sprint(v...))
}
// Noticef logs a notice using plugin_log(). Arguments are handled in the
@@ -81,7 +90,7 @@ func Noticef(format string, v ...interface{}) error {
// Info logs a purely informal message using plugin_log(). Arguments are
// handled in the manner of fmt.Print.
func Info(v ...interface{}) error {
- return log(logInfo, fmt.Sprint(v...))
+ return log(SeverityInfo, fmt.Sprint(v...))
}
// Infof logs a purely informal message using plugin_log(). Arguments are
@@ -93,7 +102,7 @@ func Infof(format string, v ...interface{}) error {
// Debug logs a debugging message using plugin_log(). Arguments are handled in
// the manner of fmt.Print.
func Debug(v ...interface{}) error {
- return log(logDebug, fmt.Sprint(v...))
+ return log(SeverityDebug, fmt.Sprint(v...))
}
// Debugf logs a debugging message using plugin_log(). Arguments are handled in
@@ -101,3 +110,14 @@ func Debug(v ...interface{}) error {
func Debugf(format string, v ...interface{}) error {
return Debug(fmt.Sprintf(format, v...))
}
+
+// LogWriter implements the io.Writer interface on top of collectd's logging facility.
+type LogWriter Severity
+
+// Write converts p to a string and logs it with w's severity.
+func (w LogWriter) Write(p []byte) (n int, err error) {
+ if err := log(Severity(w), string(p)); err != nil {
+ return 0, err
+ }
+ return len(p), nil
+}
diff --git a/plugin/log_test.go b/plugin/log_test.go
new file mode 100644
index 0000000..3e0702d
--- /dev/null
+++ b/plugin/log_test.go
@@ -0,0 +1,21 @@
+package plugin_test
+
+import (
+ "errors"
+ "log"
+ "net/http"
+
+ "collectd.org/plugin"
+)
+
+func ExampleLogWriter() {
+ l := log.New(plugin.LogWriter(plugin.SeverityError), "", log.Lshortfile)
+
+ // Start an HTTP server that logs errors to collectd's logging facility.
+ srv := &http.Server{
+ ErrorLog: l,
+ }
+ if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
+ l.Println("ListenAndServe:", err)
+ }
+}
diff --git a/plugin/plugin.go b/plugin/plugin.go
index da03209..a312ced 100644
--- a/plugin/plugin.go
+++ b/plugin/plugin.go
@@ -14,35 +14,37 @@ example:
package main
import (
- "time"
+ "context"
+ "fmt"
+ "time"
- "collectd.org/api"
- "collectd.org/plugin"
+ "collectd.org/api"
+ "collectd.org/plugin"
)
- type ExamplePlugin struct{}
-
- func (*ExamplePlugin) Read() error {
- vl := &api.ValueList{
- Identifier: api.Identifier{
- Host: "example.com",
- Plugin: "goplug",
- Type: "gauge",
- },
- Time: time.Now(),
- Interval: 10 * time.Second,
- Values: []api.Value{api.Gauge(42)},
- DSNames: []string{"value"},
- }
- if err := plugin.Write(context.Background(), vl); err != nil {
- return err
- }
-
- return nil
+ type examplePlugin struct{}
+
+ func (examplePlugin) Read(ctx context.Context) error {
+ vl := &api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "goplug",
+ Type: "gauge",
+ },
+ Time: time.Now(),
+ Interval: 10 * time.Second,
+ Values: []api.Value{api.Gauge(42)},
+ DSNames: []string{"value"},
+ }
+ if err := plugin.Write(ctx, vl); err != nil {
+ return fmt.Errorf("plugin.Write: %w", err)
+ }
+
+ return nil
}
func init() {
- plugin.RegisterRead("example", &ExamplePlugin{})
+ plugin.RegisterRead("example", examplePlugin{})
}
func main() {} // ignored
@@ -55,8 +57,8 @@ function in C based plugins.
Then, define a type which implements the Reader interface by implementing the
"Read() error" function. In the example above, this type is called
-ExamplePlugin. Create an instance of this type and pass it to RegisterRead() in
-the init() function.
+"examplePlugin". Create an instance of this type and pass it to RegisterRead()
+in the init() function.
Build flags
@@ -72,14 +74,13 @@ package plugin // import "collectd.org/plugin"
// #cgo CPPFLAGS: -DHAVE_CONFIG_H
// #cgo LDFLAGS: -ldl
// #include <stdlib.h>
+// #include <stdbool.h>
// #include <dlfcn.h>
// #include "plugin.h"
//
-// int dispatch_values_wrapper (value_list_t const *vl);
-// int register_read_wrapper (char const *group, char const *name,
-// plugin_read_cb callback,
-// cdtime_t interval,
-// user_data_t *ud);
+// int plugin_dispatch_values_wrapper(value_list_t const *vl);
+// cdtime_t plugin_get_interval_wrapper(void);
+// int timeout_wrapper(void);
//
// data_source_t *ds_dsrc(data_set_t const *ds, size_t i);
//
@@ -90,32 +91,69 @@ package plugin // import "collectd.org/plugin"
// derive_t value_list_get_derive (value_list_t *, size_t);
// gauge_t value_list_get_gauge (value_list_t *, size_t);
//
+// meta_data_t *meta_data_create_wrapper(void);
+// void meta_data_destroy_wrapper(meta_data_t *md);
+// int meta_data_add_boolean_wrapper(meta_data_t *md, const char *key, bool value);
+// int meta_data_add_double_wrapper(meta_data_t *md, const char *key, double value);
+// int meta_data_add_signed_int_wrapper(meta_data_t *md, const char *key, int64_t value);
+// int meta_data_add_string_wrapper(meta_data_t *md, const char *key, const char *value);
+// int meta_data_add_unsigned_int_wrapper(meta_data_t *md, const char *key, uint64_t value);
+// int meta_data_get_boolean_wrapper(meta_data_t *md, const char *key, bool *value);
+// int meta_data_get_double_wrapper(meta_data_t *md, const char *key, double *value);
+// int meta_data_get_signed_int_wrapper(meta_data_t *md, const char *key, int64_t *value);
+// int meta_data_get_string_wrapper(meta_data_t *md, const char *key, char **value);
+// int meta_data_get_unsigned_int_wrapper(meta_data_t *md, const char *key, uint64_t *value);
+// int meta_data_toc_wrapper(meta_data_t *md, char ***toc);
+// int meta_data_type_wrapper(meta_data_t *md, char const *key);
+//
+// int plugin_register_complex_read_wrapper(char const *group, char const *name,
+// plugin_read_cb callback,
+// cdtime_t interval, user_data_t *ud);
// int wrap_read_callback(user_data_t *);
//
-// int register_write_wrapper (char const *, plugin_write_cb, user_data_t *);
+// int plugin_register_write_wrapper(char const *, plugin_write_cb, user_data_t *);
// int wrap_write_callback(data_set_t *, value_list_t *, user_data_t *);
//
-// int register_shutdown_wrapper (char *, plugin_shutdown_cb);
+// int plugin_register_shutdown_wrapper(char *, plugin_shutdown_cb);
// int wrap_shutdown_callback(void);
+//
+// int plugin_register_log_wrapper(char const *, plugin_log_cb,
+// user_data_t const *);
+// int wrap_log_callback(int, char *, user_data_t *);
+//
+// typedef int (*plugin_complex_config_cb)(oconfig_item_t *);
+//
+// int register_complex_config_wrapper(char const *, plugin_complex_config_cb);
+// int wrap_configure_callback(oconfig_item_t *);
+// int dispatch_configurations(void);
+//
+// int register_init_wrapper (const char *name, plugin_init_cb callback);
+//
+// typedef void (*free_func_t)(void *);
import "C"
import (
"context"
+ "errors"
"fmt"
+ "strings"
+ "sync"
+ "time"
"unsafe"
"collectd.org/api"
"collectd.org/cdtime"
-)
-
-var (
- ctx = context.Background()
+ "collectd.org/config"
+ "collectd.org/meta"
)
// Reader defines the interface for read callbacks, i.e. Go functions that are
// called periodically from the collectd daemon.
+// The context passed to the Read() function has a timeout based on collectd's
+// "Timeout" global config option. It defaults to twice the plugin's read
+// interval.
type Reader interface {
- Read() error
+ Read(ctx context.Context) error
}
func strcpy(dst []C.char, src string) {
@@ -145,55 +183,216 @@ func newValueListT(vl *api.ValueList) (*C.value_list_t, error) {
switch v := v.(type) {
case api.Counter:
if _, err := C.value_list_add_counter(ret, C.counter_t(v)); err != nil {
- return nil, fmt.Errorf("value_list_add_counter: %v", err)
+ return nil, fmt.Errorf("value_list_add_counter: %w", err)
}
case api.Derive:
if _, err := C.value_list_add_derive(ret, C.derive_t(v)); err != nil {
- return nil, fmt.Errorf("value_list_add_derive: %v", err)
+ return nil, fmt.Errorf("value_list_add_derive: %w", err)
}
case api.Gauge:
if _, err := C.value_list_add_gauge(ret, C.gauge_t(v)); err != nil {
- return nil, fmt.Errorf("value_list_add_gauge: %v", err)
+ return nil, fmt.Errorf("value_list_add_gauge: %w", err)
}
default:
return nil, fmt.Errorf("not yet supported: %T", v)
}
}
+ md, err := marshalMeta(vl.Meta)
+ if err != nil {
+ return nil, err
+ }
+ ret.meta = md
+
return ret, nil
}
-// writer implements the api.Write interface.
-type writer struct{}
+func freeValueListT(vl *C.value_list_t) {
+ C.free(unsafe.Pointer(vl.values))
+ vl.values = nil
+ if vl.meta != nil {
+ C.meta_data_destroy_wrapper(vl.meta)
+ vl.meta = nil
+ }
+}
+
+func marshalMeta(meta meta.Data) (*C.meta_data_t, error) {
+ if meta == nil {
+ return nil, nil
+ }
+
+ md, err := C.meta_data_create_wrapper()
+ if err != nil {
+ return nil, wrapCError(0, err, "meta_data_create")
+ }
+
+ for k, v := range meta {
+ if err := marshalMetaEntry(md, k, v); err != nil {
+ C.meta_data_destroy_wrapper(md)
+ return nil, err
+ }
+ }
+
+ return md, nil
+}
+
+func marshalMetaEntry(md *C.meta_data_t, key string, value meta.Entry) error {
+ cKey := C.CString(key)
+ defer C.free(unsafe.Pointer(cKey))
+
+ switch value := value.Interface().(type) {
+ case bool:
+ s, err := C.meta_data_add_boolean_wrapper(md, cKey, C.bool(value))
+ return wrapCError(s, err, "meta_data_add_boolean")
+ case float64:
+ s, err := C.meta_data_add_double_wrapper(md, cKey, C.double(value))
+ return wrapCError(s, err, "meta_data_add_double")
+ case int64:
+ s, err := C.meta_data_add_signed_int_wrapper(md, cKey, C.int64_t(value))
+ return wrapCError(s, err, "meta_data_add_signed_int")
+ case uint64:
+ s, err := C.meta_data_add_unsigned_int_wrapper(md, cKey, C.uint64_t(value))
+ return wrapCError(s, err, "meta_data_add_unsigned_int")
+ case string:
+ cValue := C.CString(value)
+ defer C.free(unsafe.Pointer(cValue))
+ s, err := C.meta_data_add_string_wrapper(md, cKey, cValue)
+ return wrapCError(s, err, "meta_data_add_string")
+ default:
+ return nil
+ }
+}
-// NewWriter returns an object implementing the api.Writer interface for the
-// collectd daemon.
-func NewWriter() api.Writer {
- return writer{}
+// cStrarrayIndex returns the n'th string in the array, i.e. strings[n].
+func cStrarrayIndex(strings **C.char, n int) *C.char {
+ offset := uintptr(n) * unsafe.Sizeof(*strings)
+ ptr := (**C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(strings)) + offset))
+ return *ptr
}
-// Write implements the api.Writer interface for the collectd daemon.
-func (writer) Write(_ context.Context, vl *api.ValueList) error {
- return Write(vl)
+func unmarshalMeta(md *C.meta_data_t) (meta.Data, error) {
+ if md == nil {
+ return nil, nil
+ }
+
+ var ptr **C.char
+ num, err := C.meta_data_toc_wrapper(md, &ptr)
+ if num < 0 || err != nil {
+ return nil, wrapCError(num, err, "meta_data_toc")
+ }
+ if num < 1 {
+ return nil, nil
+ }
+ defer func() {
+ for i := 0; i < int(num); i++ {
+ C.free(unsafe.Pointer(cStrarrayIndex(ptr, i)))
+ }
+ C.free(unsafe.Pointer(ptr))
+ }()
+
+ ret := make(meta.Data)
+ for i := 0; i < int(num); i++ {
+ key := cStrarrayIndex(ptr, i)
+ if err := unmarshalMetaEntry(ret, md, key); err != nil {
+ return nil, err
+ }
+ }
+
+ return ret, nil
+}
+
+func unmarshalMetaEntry(goMeta meta.Data, cMeta *C.meta_data_t, key *C.char) error {
+ typ, err := C.meta_data_type_wrapper(cMeta, key)
+ if typ <= 0 || err != nil {
+ if typ == 0 && err == nil {
+ err = fmt.Errorf("no such meta data key: %q", C.GoString(key))
+ }
+ return wrapCError(typ, err, "meta_data_type")
+ }
+
+ switch typ {
+ case C.MD_TYPE_BOOLEAN:
+ var v C.bool
+ s, err := C.meta_data_get_boolean_wrapper(cMeta, key, &v)
+ if err := wrapCError(s, err, "meta_data_get_boolean"); err != nil {
+ return err
+ }
+ goMeta[C.GoString(key)] = meta.Bool(bool(v))
+ case C.MD_TYPE_DOUBLE:
+ var v C.double
+ s, err := C.meta_data_get_double_wrapper(cMeta, key, &v)
+ if err := wrapCError(s, err, "meta_data_get_double"); err != nil {
+ return err
+ }
+ goMeta[C.GoString(key)] = meta.Float64(float64(v))
+ case C.MD_TYPE_SIGNED_INT:
+ var v C.int64_t
+ s, err := C.meta_data_get_signed_int_wrapper(cMeta, key, &v)
+ if err := wrapCError(s, err, "meta_data_get_signed_int"); err != nil {
+ return err
+ }
+ goMeta[C.GoString(key)] = meta.Int64(int64(v))
+ case C.MD_TYPE_STRING:
+ var v *C.char
+ s, err := C.meta_data_get_string_wrapper(cMeta, key, &v)
+ if err := wrapCError(s, err, "meta_data_get_string"); err != nil {
+ return err
+ }
+ defer C.free(unsafe.Pointer(v))
+ goMeta[C.GoString(key)] = meta.String(C.GoString(v))
+ case C.MD_TYPE_UNSIGNED_INT:
+ var v C.uint64_t
+ s, err := C.meta_data_get_unsigned_int_wrapper(cMeta, key, &v)
+ if err := wrapCError(s, err, "meta_data_get_unsigned_int"); err != nil {
+ return err
+ }
+ goMeta[C.GoString(key)] = meta.UInt64(uint64(v))
+ default:
+ Warningf("unexpected meta data type %v", typ)
+ }
+
+ return nil
}
// Write converts a ValueList and calls the plugin_dispatch_values() function
// of the collectd daemon.
-func Write(vl *api.ValueList) error {
- vlt, err := newValueListT(vl)
- if err != nil {
- return err
+//
+// The following fields are optional and will be filled in if empty / zero:
+//
+// · vl.Identifier.Host
+//
+// · vl.Identifier.Plugin
+//
+// · vl.Time
+//
+// · vl.Interval
+//
+// Use api.WriterFunc to pass this function as an api.Writer.
+func Write(ctx context.Context, vl *api.ValueList) error {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ default:
+ }
+
+ if vl.Plugin == "" {
+ n, ok := Name(ctx)
+ if !ok {
+ return errors.New("unable to determine plugin name from context")
+ }
+ // Don't modify the argument.
+ vl = vl.Clone()
+ vl.Plugin = n
}
- defer C.free(unsafe.Pointer(vlt.values))
- status, err := C.dispatch_values_wrapper(vlt)
+ vlt, err := newValueListT(vl)
if err != nil {
return err
- } else if status != 0 {
- return fmt.Errorf("dispatch_values failed with status %d", status)
}
+ defer freeValueListT(vlt)
- return nil
+ status, err := C.plugin_dispatch_values_wrapper(vlt)
+ return wrapCError(status, err, "plugin_dispatch_values")
}
// readFuncs holds references to all read callbacks, so the garbage collector
@@ -202,30 +401,80 @@ var readFuncs = make(map[string]Reader)
// RegisterRead registers a new read function with the daemon which is called
// periodically.
-func RegisterRead(name string, r Reader) error {
- cGroup := C.CString("golang")
- defer C.free(unsafe.Pointer(cGroup))
+func RegisterRead(name string, r Reader, opts ...ReadOption) error {
+ ro := readOpt{
+ group: "golang",
+ }
+
+ for _, opt := range opts {
+ opt(&ro)
+ }
+
+ var cGroup *C.char
+ if ro.group != "" {
+ cGroup = C.CString(ro.group)
+ defer C.free(unsafe.Pointer(cGroup))
+ }
cName := C.CString(name)
ud := C.user_data_t{
data: unsafe.Pointer(cName),
- free_func: nil,
+ free_func: C.free_func_t(C.free),
}
- status, err := C.register_read_wrapper(cGroup, cName,
+ status, err := C.plugin_register_complex_read_wrapper(cGroup, cName,
C.plugin_read_cb(C.wrap_read_callback),
- C.cdtime_t(0),
+ C.cdtime_t(ro.interval),
&ud)
- if err != nil {
+ if err := wrapCError(status, err, "plugin_register_complex_read"); err != nil {
return err
- } else if status != 0 {
- return fmt.Errorf("register_read_wrapper failed with status %d", status)
}
readFuncs[name] = r
return nil
}
+type readOpt struct {
+ group string
+ interval cdtime.Time
+}
+
+// ReadOption is an option for the RegisterRead function.
+type ReadOption func(o *readOpt)
+
+// WithInterval sets the interval in which the read callback is being called.
+// If unspecified, or when set to zero, collectd's global default is used.
+//
+// The vast majority of plugins SHOULD NOT set this option explicitly and
+// respect the user's configuration by using the default instead.
+func WithInterval(d time.Duration) ReadOption {
+ return func(o *readOpt) {
+ o.interval = cdtime.NewDuration(d)
+ }
+}
+
+// WithGroup sets the group name of the read callback. If unspecified, "golang"
+// is used. Set to the empty string to clear the group name.
+func WithGroup(g string) ReadOption {
+ return func(o *readOpt) {
+ o.group = g
+ }
+}
+
+type key struct{}
+
+var nameKey key
+
+func withName(ctx context.Context, name string) context.Context {
+ return context.WithValue(ctx, nameKey, name)
+}
+
+// Name returns the name of the plugin / callback.
+func Name(ctx context.Context) (string, bool) {
+ name, ok := ctx.Value(nameKey).(string)
+ return name, ok
+}
+
//export wrap_read_callback
func wrap_read_callback(ud *C.user_data_t) C.int {
name := C.GoString((*C.char)(ud.data))
@@ -234,7 +483,18 @@ func wrap_read_callback(ud *C.user_data_t) C.int {
return -1
}
- if err := r.Read(); err != nil {
+ timeout, err := Timeout()
+ if err != nil {
+ Errorf("%s plugin: Timeout() failed: %v", name, err)
+ return -1
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), timeout)
+ defer cancel()
+
+ ctx = withName(ctx, name)
+
+ if err := r.Read(ctx); err != nil {
Errorf("%s plugin: Read() failed: %v", name, err)
return -1
}
@@ -242,6 +502,32 @@ func wrap_read_callback(ud *C.user_data_t) C.int {
return 0
}
+// Interval returns the interval in which read callbacks are being called. May
+// only be called from within a read callback.
+func Interval() (time.Duration, error) {
+ ival, err := C.plugin_get_interval_wrapper()
+ if err != nil {
+ return 0, fmt.Errorf("plugin_get_interval() failed: %w", err)
+ }
+
+ return cdtime.Time(ival).Duration(), nil
+}
+
+// Timeout returns the duration after which this plugin's metrics are
+// considered stale and are pruned from collectd's internal metrics cache.
+func Timeout() (time.Duration, error) {
+ to, err := C.timeout_wrapper()
+ if err != nil {
+ return 0, fmt.Errorf("timeout_wrapper() failed: %w", err)
+ }
+ ival, err := Interval()
+ if err != nil {
+ return 0, err
+ }
+
+ return ival * time.Duration(to), nil
+}
+
// writeFuncs holds references to all write callbacks, so the garbage collector
// doesn't get any funny ideas.
var writeFuncs = make(map[string]api.Writer)
@@ -256,14 +542,12 @@ func RegisterWrite(name string, w api.Writer) error {
cName := C.CString(name)
ud := C.user_data_t{
data: unsafe.Pointer(cName),
- free_func: nil,
+ free_func: C.free_func_t(C.free),
}
- status, err := C.register_write_wrapper(cName, C.plugin_write_cb(C.wrap_write_callback), &ud)
- if err != nil {
+ status, err := C.plugin_register_write_wrapper(cName, C.plugin_write_cb(C.wrap_write_callback), &ud)
+ if err := wrapCError(status, err, "plugin_register_write"); err != nil {
return err
- } else if status != 0 {
- return fmt.Errorf("register_write_wrapper failed with status %d", status)
}
writeFuncs[name] = w
@@ -312,6 +596,13 @@ func wrap_write_callback(ds *C.data_set_t, cvl *C.value_list_t, ud *C.user_data_
vl.DSNames = append(vl.DSNames, C.GoString(&dsrc.name[0]))
}
+ m, err := unmarshalMeta(cvl.meta)
+ if err != nil {
+ Errorf("%s plugin: unmarshalMeta() failed: %v", name, err)
+ }
+ vl.Meta = m
+
+ ctx := withName(context.Background(), name)
if err := w.Write(ctx, vl); err != nil {
Errorf("%s plugin: Write() failed: %v", name, err)
return -1
@@ -322,9 +613,9 @@ func wrap_write_callback(ds *C.data_set_t, cvl *C.value_list_t, ud *C.user_data_
// First declare some types, interfaces, general functions
-// Shutters are objects that when called will shut down the plugin gracefully
+// Shutter is called to shut down the plugin gracefully.
type Shutter interface {
- Shutdown() error
+ Shutdown(context.Context) error
}
// shutdownFuncs holds references to all shutdown callbacks
@@ -332,30 +623,29 @@ var shutdownFuncs = make(map[string]Shutter)
//export wrap_shutdown_callback
func wrap_shutdown_callback() C.int {
- if len(shutdownFuncs) <= 0 {
- return 0
- }
- for n, s := range shutdownFuncs {
- if err := s.Shutdown(); err != nil {
- Errorf("%s plugin: Shutdown() failed: %v", n, s)
- return -1
+ ret := C.int(0)
+ for name, f := range shutdownFuncs {
+ ctx := withName(context.Background(), name)
+ if err := f.Shutdown(ctx); err != nil {
+ Errorf("%s plugin: Shutdown() failed: %v", name, err)
+ ret = -1
}
}
- return 0
+ return ret
}
// RegisterShutdown registers a shutdown function with the daemon which is called
// when the plugin is required to shutdown gracefully.
func RegisterShutdown(name string, s Shutter) error {
// Only register a callback the first time one is implemented, subsequent
- // callbacks get added to a list and called at the same time
+ // callbacks get added to a map and called sequentially from the same
+ // (C) callback.
if len(shutdownFuncs) <= 0 {
cName := C.CString(name)
- cCallback := C.plugin_shutdown_cb(C.wrap_shutdown_callback)
+ defer C.free(unsafe.Pointer(cName))
- status, err := C.register_shutdown_wrapper(cName, cCallback)
- if err != nil {
- Errorf("register_shutdown_wrapper failed with status: %v", status)
+ status, err := C.plugin_register_shutdown_wrapper(cName, C.plugin_shutdown_cb(C.wrap_shutdown_callback))
+ if err := wrapCError(status, err, "plugin_register_shutdown"); err != nil {
return err
}
}
@@ -363,6 +653,151 @@ func RegisterShutdown(name string, s Shutter) error {
return nil
}
+// Logger implements a logging callback.
+type Logger interface {
+ Log(context.Context, Severity, string)
+}
+
+// RegisterLog registers a logging function with the daemon which is called
+// whenever a log message is generated.
+func RegisterLog(name string, l Logger) error {
+ cName := C.CString(name)
+ ud := C.user_data_t{
+ data: unsafe.Pointer(cName),
+ free_func: C.free_func_t(C.free),
+ }
+
+ status, err := C.plugin_register_log_wrapper(cName, C.plugin_log_cb(C.wrap_log_callback), &ud)
+ if err := wrapCError(status, err, "plugin_register_log"); err != nil {
+ return err
+ }
+
+ logFuncs[name] = l
+ return nil
+}
+
+var logFuncs = make(map[string]Logger)
+
+//export wrap_log_callback
+func wrap_log_callback(sev C.int, msg *C.char, ud *C.user_data_t) C.int {
+ name := C.GoString((*C.char)(ud.data))
+ f, ok := logFuncs[name]
+ if !ok {
+ return -1
+ }
+
+ ctx := withName(context.Background(), name)
+ f.Log(ctx, Severity(sev), C.GoString(msg))
+
+ return 0
+}
+
+// Configurer implements a Configure callback.
+type Configurer interface {
+ Configure(context.Context, config.Block) error
+}
+
+// Configurers are registered once but Configs may be received multiple times
+// and merged together before unmarshalling, so they're tracked together for a
+// convenient Unmarshal call.
+type configFunc struct {
+ Configurer
+ cfg config.Block
+}
+
+var (
+ configureFuncs = make(map[string]*configFunc)
+ registerConfigInit sync.Once
+)
+
+// RegisterConfig registers a configuration-receiving function with the daemon.
+//
+// c.Configure is called exactly once after the entire configuration has been
+// read. If there are multiple configuration blocks for the plugin, they will
+// be merged into a single block using "collectd.org/config".Block.Merge.
+//
+// If no configuration is found for "name", c.Configure is still called with a
+// zero-valued config.Block.
+func RegisterConfig(name string, c Configurer) error {
+ cName := C.CString(name)
+ defer C.free(unsafe.Pointer(cName))
+
+ var regErr error
+ registerConfigInit.Do(func() {
+ status, err := C.register_init_wrapper(cName, C.plugin_init_cb(C.dispatch_configurations))
+ regErr = wrapCError(status, err, "plugin_register_init")
+ })
+ if regErr != nil {
+ return regErr
+ }
+
+ status, err := C.register_complex_config_wrapper(cName, C.plugin_complex_config_cb(C.wrap_configure_callback))
+ if err := wrapCError(status, err, "register_configure"); err != nil {
+ return err
+ }
+
+ configureFuncs[name] = &configFunc{
+ Configurer: c,
+ }
+ return nil
+}
+
+//export wrap_configure_callback
+func wrap_configure_callback(ci *C.oconfig_item_t) C.int {
+ block, err := unmarshalConfigBlock(ci)
+ if err != nil {
+ Errorf("unmarshalConfigBlock: %v", err)
+ return -1
+ }
+
+ key := strings.ToLower(block.Key)
+ if key != "plugin" {
+ Errorf("got config block %q, want %q", block.Key, "Plugin")
+ return -1
+ }
+ block.Key = key
+
+ if len(block.Values) != 1 || !block.Values[0].IsString() {
+ Errorf("got Values=%v, want single string value", block)
+ return -1
+ }
+ plugin := block.Values[0].String()
+
+ f, ok := configureFuncs[plugin]
+ if !ok {
+ Errorf("callback for plugin %q not found", plugin)
+ return -1
+ }
+
+ if err := f.cfg.Merge(block); err != nil {
+ Errorf("merging config blocks failed: %v", err)
+ return -1
+ }
+
+ return 0
+}
+
+//export dispatch_configurations
+func dispatch_configurations() C.int {
+ for name, f := range configureFuncs {
+ ctx := withName(context.Background(), name)
+ if err := f.Configure(ctx, f.cfg); err != nil {
+ Errorf("%s plugin: Configure() failed: %v", name, err)
+ }
+ }
+ return 0
+}
+
+func wrapCError(status C.int, err error, name string) error {
+ if err != nil {
+ return fmt.Errorf("%s failed: %w", name, err)
+ }
+ if status != 0 {
+ return fmt.Errorf("%s failed with status %d", name, status)
+ }
+ return nil
+}
+
//export module_register
func module_register() {
}
diff --git a/plugin/plugin_test.go b/plugin/plugin_test.go
new file mode 100644
index 0000000..0d238e3
--- /dev/null
+++ b/plugin/plugin_test.go
@@ -0,0 +1,454 @@
+package plugin_test
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "math"
+ "testing"
+ "time"
+
+ "collectd.org/api"
+ "collectd.org/meta"
+ "collectd.org/plugin"
+ "collectd.org/plugin/fake"
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+)
+
+func TestInterval(t *testing.T) {
+ fake.SetInterval(42 * time.Second)
+ defer fake.TearDown()
+
+ got, err := plugin.Interval()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if want := 42 * time.Second; got != want {
+ t.Errorf("Interval() = %v, want %v", got, want)
+ }
+}
+
+type testLogger struct {
+ Name string
+ plugin.Severity
+ Message string
+}
+
+func (l *testLogger) Log(ctx context.Context, s plugin.Severity, msg string) {
+ l.Severity = s
+ l.Message = msg
+ l.Name, _ = plugin.Name(ctx)
+}
+
+func (l *testLogger) reset() {
+ *l = testLogger{}
+}
+
+func TestLog(t *testing.T) {
+ cases := []struct {
+ title string
+ logFunc func(v ...interface{}) error
+ fmtFunc func(format string, v ...interface{}) error
+ severity plugin.Severity
+ }{
+ {"Error", plugin.Error, plugin.Errorf, plugin.SeverityError},
+ {"Warning", plugin.Warning, plugin.Warningf, plugin.SeverityWarning},
+ {"Notice", plugin.Notice, plugin.Noticef, plugin.SeverityNotice},
+ {"Info", plugin.Info, plugin.Infof, plugin.SeverityInfo},
+ {"Debug", plugin.Debug, plugin.Debugf, plugin.SeverityDebug},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ defer fake.TearDown()
+
+ name := "TestLog_" + tc.title
+ l := &testLogger{}
+ if err := plugin.RegisterLog(name, l); err != nil {
+ t.Fatal(err)
+ }
+
+ tc.logFunc("test %d %%s", 42)
+ if got, want := l.Name, name; got != want {
+ t.Errorf("plugin.Name() = %q, want %q", got, want)
+ }
+ if got, want := l.Severity, tc.severity; got != want {
+ t.Errorf("Severity = %v, want %v", got, want)
+ }
+ if got, want := l.Message, "test %d %%s42"; got != want {
+ t.Errorf("Message = %q, want %q", got, want)
+ }
+
+ l.reset()
+ tc.fmtFunc("test %d %%s", 42)
+ if got, want := l.Name, name; got != want {
+ t.Errorf("plugin.Name() = %q, want %q", got, want)
+ }
+ if got, want := l.Severity, tc.severity; got != want {
+ t.Errorf("Severity = %v, want %v", got, want)
+ }
+ if got, want := l.Message, "test 42 %s"; got != want {
+ t.Errorf("Message = %q, want %q", got, want)
+ }
+
+ l.reset()
+ fmt.Fprintln(plugin.LogWriter(tc.severity), "test message:", 42)
+ if got, want := l.Name, name; got != want {
+ t.Errorf("plugin.Name() = %q, want %q", got, want)
+ }
+ if got, want := l.Severity, tc.severity; got != want {
+ t.Errorf("Severity = %v, want %v", got, want)
+ }
+ if got, want := l.Message, "test message: 42"; got != want {
+ t.Errorf("Message = %q, want %q", got, want)
+ }
+ })
+ }
+}
+
+func TestRegisterRead(t *testing.T) {
+ cases := []struct {
+ title string
+ opts []plugin.ReadOption
+ wantGroup string
+ wantInterval time.Duration
+ }{
+ {
+ title: "default case",
+ wantGroup: "golang",
+ wantInterval: 10 * time.Second,
+ },
+ {
+ title: "with interval",
+ opts: []plugin.ReadOption{plugin.WithInterval(20 * time.Second)},
+ wantGroup: "golang",
+ wantInterval: 20 * time.Second,
+ },
+ {
+ title: "with group",
+ opts: []plugin.ReadOption{plugin.WithGroup("testing")},
+ wantGroup: "testing",
+ wantInterval: 10 * time.Second,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ defer fake.TearDown()
+
+ if err := plugin.RegisterRead("TestRegisterRead", &testReader{}, tc.opts...); err != nil {
+ t.Fatal(err)
+ }
+
+ callbacks := fake.ReadCallbacks()
+ if got, want := len(callbacks), 1; got != want {
+ t.Errorf("len(ReadCallbacks) = %d, want %d", got, want)
+ }
+ if len(callbacks) < 1 {
+ t.FailNow()
+ }
+
+ cb := callbacks[0]
+ if got, want := cb.Group, tc.wantGroup; got != want {
+ t.Errorf("ReadCallback.Group = %q, want %q", got, want)
+ }
+ if got, want := cb.Interval.Duration(), tc.wantInterval; got != want {
+ t.Errorf("ReadCallback.Interval = %v, want %v", got, want)
+ }
+ })
+ }
+}
+
+func TestReadWrite(t *testing.T) {
+ baseVL := api.ValueList{
+ Identifier: api.Identifier{
+ Host: "example.com",
+ Plugin: "TestRead",
+ Type: "gauge",
+ },
+ Time: time.Unix(1587500000, 0),
+ Interval: 10 * time.Second,
+ Values: []api.Value{api.Gauge(42)},
+ DSNames: []string{"value"},
+ }
+
+ cases := []struct {
+ title string
+ modifyVL func(*api.ValueList)
+ readErr error
+ writeErr error
+ wantErr bool
+ }{
+ {
+ title: "gauge",
+ },
+ {
+ title: "gauge NaN",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Values = []api.Value{api.Gauge(math.NaN())}
+ },
+ },
+ {
+ title: "derive",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Type = "derive"
+ vl.Values = []api.Value{api.Derive(42)}
+ },
+ },
+ {
+ title: "counter",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Type = "counter"
+ vl.Values = []api.Value{api.Counter(42)}
+ },
+ },
+ {
+ title: "bool meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.Bool(true),
+ }
+ },
+ },
+ {
+ title: "float64 meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.Float64(20.0 / 3.0),
+ }
+ },
+ },
+ {
+ title: "float64 NaN meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.Float64(math.NaN()),
+ }
+ },
+ },
+ {
+ title: "int64 meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.Int64(-23),
+ }
+ },
+ },
+ {
+ title: "uint64 meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.UInt64(42),
+ }
+ },
+ },
+ {
+ title: "string meta data",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Meta = meta.Data{
+ "key": meta.String(`\\\ value ///`),
+ }
+ },
+ },
+ {
+ title: "marshaling error",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Values = []api.Value{nil}
+ },
+ wantErr: true,
+ },
+ {
+ title: "read callback sets errno",
+ // The "plugin_dispatch_values()" implementation of the "fake" package only supports the types
+ // "derive", "gauge", and "counter". If another type is encountered, errno is set to EINVAL.
+ modifyVL: func(vl *api.ValueList) {
+ vl.Type = "invalid"
+ },
+ wantErr: true,
+ },
+ {
+ title: "read callback returns error",
+ readErr: errors.New("read error"),
+ wantErr: true,
+ },
+ {
+ title: "read callback canceled context",
+ // Calling plugin.Write() with a canceled context results in an error.
+ readErr: context.Canceled,
+ wantErr: true,
+ },
+ {
+ title: "write callback returns error",
+ writeErr: errors.New("write error"),
+ wantErr: true,
+ },
+ {
+ title: "plugin name is filled in",
+ modifyVL: func(vl *api.ValueList) {
+ vl.Plugin = ""
+ },
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.title, func(t *testing.T) {
+ defer fake.TearDown()
+
+ vl := baseVL.Clone()
+ if tc.modifyVL != nil {
+ tc.modifyVL(vl)
+ }
+
+ r := &testReader{
+ vl: vl,
+ wantName: "TestRead",
+ wantErr: tc.readErr,
+ }
+ if err := plugin.RegisterRead("TestRead", r); err != nil {
+ t.Fatal(err)
+ }
+
+ w := &testWriter{
+ wantName: "TestWrite",
+ wantErr: tc.writeErr,
+ }
+ if err := plugin.RegisterWrite("TestWrite", w); err != nil {
+ t.Fatal(err)
+ }
+
+ err := fake.ReadAll()
+ if gotErr := err != nil; gotErr != tc.wantErr {
+ t.Errorf("ReadAll() = %v, want error: %v", err, tc.wantErr)
+ }
+ if tc.wantErr {
+ return
+ }
+
+ if got, want := len(w.valueLists), 1; got != want {
+ t.Errorf("len(testWriter.valueLists) = %d, want %d", got, want)
+ }
+ if len(w.valueLists) < 1 {
+ t.FailNow()
+ }
+
+ // Expect vl.Plugin to get populated.
+ if vl.Plugin == "" {
+ vl.Plugin = "TestRead"
+ }
+
+ opts := []cmp.Option{
+ // cmp complains about meta.Entry having private fields.
+ cmp.Transformer("meta.Entry", func(e meta.Entry) interface{} {
+ return e.Interface()
+ }),
+ // transform api.Gauge to float64, so EquateNaNs applies to them.
+ cmp.Transformer("api.Gauge", func(g api.Gauge) interface{} {
+ return float64(g)
+ }),
+ cmpopts.EquateNaNs(),
+ }
+ if got, want := w.valueLists[0], vl; !cmp.Equal(got, want, opts...) {
+ t.Errorf("ValueList differs (-want/+got): %s", cmp.Diff(want, got, opts...))
+ }
+ })
+ }
+}
+
+type testReader struct {
+ vl *api.ValueList
+ wantName string
+ wantErr error
+}
+
+func (r *testReader) Read(ctx context.Context) error {
+ // Verify that plugin.Name() works inside Read callbacks.
+ gotName, ok := plugin.Name(ctx)
+ if !ok || gotName != r.wantName {
+ return fmt.Errorf("plugin.Name() = (%q, %v), want (%q, %v)", gotName, ok, r.wantName, true)
+ }
+
+ if errors.Is(r.wantErr, context.Canceled) {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithCancel(ctx)
+ cancel()
+ // continue with canceled context
+ } else if r.wantErr != nil {
+ return r.wantErr
+ }
+
+ return plugin.Write(ctx, r.vl)
+}
+
+type testWriter struct {
+ valueLists []*api.ValueList
+ wantName string
+ wantErr error
+}
+
+func (w *testWriter) Write(ctx context.Context, vl *api.ValueList) error {
+ // Verify that plugin.Name() works inside Write callbacks.
+ gotName, ok := plugin.Name(ctx)
+ if !ok || gotName != w.wantName {
+ return fmt.Errorf("plugin.Name() = (%q, %v), want (%q, %v)", gotName, ok, w.wantName, true)
+ }
+
+ if w.wantErr != nil {
+ return w.wantErr
+ }
+
+ w.valueLists = append(w.valueLists, vl)
+ return nil
+}
+
+func TestShutdown(t *testing.T) {
+ // NOTE: fake.TearDown() will remove all callbacks from the C code's state. plugin.shutdownFuncs will still hold
+ // a reference to the registered shutdown calls, preventing it from registering another C callback in later
+ // tests. Long story short, don't use shutdown callbacks in any other test.
+ defer fake.TearDown()
+
+ shutters := []*testShutter{}
+ // This creates 20 shutdown functions: one will succeed, 19 will fail.
+ // We expect *all* shutdown functions to be called.
+ for i := 0; i < 20; i++ {
+ s := &testShutter{
+ wantName: "TestShutdown",
+ }
+ callbackName := "TestShutdown"
+ if i != 0 {
+ callbackName = fmt.Sprintf("failing_function_%d", i)
+ }
+
+ if err := plugin.RegisterShutdown(callbackName, s); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ if err := fake.ShutdownAll(); err == nil {
+ t.Error("fake.ShutdownAll() succeeded, expected it to fail")
+ }
+
+ for _, s := range shutters {
+ if got, want := s.callCount, 1; got != want {
+ t.Errorf("testShutter.callCount = %d, want %d", got, want)
+ }
+ }
+}
+
+type testShutter struct {
+ wantName string
+ callCount int
+}
+
+func (s *testShutter) Shutdown(ctx context.Context) error {
+ s.callCount++
+
+ // Verify that plugin.Name() works inside Shutdown callbacks.
+ gotName, ok := plugin.Name(ctx)
+ if !ok || gotName != s.wantName {
+ return fmt.Errorf("plugin.Name() = (%q, %v), want (%q, %v)", gotName, ok, s.wantName, true)
+ }
+
+ return nil
+}
diff --git a/plugin/wrapper.go b/plugin/wrapper.go
new file mode 100644
index 0000000..fcd3cbe
--- /dev/null
+++ b/plugin/wrapper.go
@@ -0,0 +1,164 @@
+// +build go1.5,cgo
+
+package plugin // import "collectd.org/plugin"
+
+// #cgo CPPFLAGS: -DHAVE_CONFIG_H
+// #cgo LDFLAGS: -ldl
+// #include "plugin.h"
+// #include <dlfcn.h>
+// #include <stdbool.h>
+// #include <stdlib.h>
+//
+// #define LOAD(f) \
+// if (f##_ptr == NULL) { \
+// void *hnd = dlopen(NULL, RTLD_LAZY); \
+// f##_ptr = dlsym(hnd, #f); \
+// dlclose(hnd); \
+// }
+//
+// static int (*meta_data_add_boolean_ptr)(meta_data_t *, char const *, bool);
+// static int (*meta_data_add_double_ptr)(meta_data_t *, char const *, double);
+// static int (*meta_data_add_signed_int_ptr)(meta_data_t *, char const *,
+// int64_t);
+// static int (*meta_data_add_string_ptr)(meta_data_t *, char const *,
+// char const *);
+// static int (*meta_data_add_unsigned_int_ptr)(meta_data_t *, char const *,
+// uint64_t);
+// static meta_data_t *(*meta_data_create_ptr)(void);
+// static void (*meta_data_destroy_ptr)(meta_data_t *);
+// static int (*meta_data_get_boolean_ptr)(meta_data_t *, char const *, bool *);
+// static int (*meta_data_get_double_ptr)(meta_data_t *, char const *, double *);
+// static int (*meta_data_get_signed_int_ptr)(meta_data_t *, char const *,
+// int64_t *);
+// static int (*meta_data_get_string_ptr)(meta_data_t *, char const *, char **);
+// static int (*meta_data_get_unsigned_int_ptr)(meta_data_t *, char const *,
+// uint64_t *);
+// static int (*meta_data_toc_ptr)(meta_data_t *, char ***);
+// static int (*meta_data_type_ptr)(meta_data_t *, char const *);
+// static int (*plugin_dispatch_values_ptr)(value_list_t const *);
+// static cdtime_t (*plugin_get_interval_ptr)(void);
+// static int (*plugin_register_complex_read_ptr)(meta_data_t *, char const *,
+// plugin_read_cb, cdtime_t,
+// user_data_t *);
+// static int (*plugin_register_log_ptr)(char const *, plugin_log_cb,
+// user_data_t *);
+// static int (*plugin_register_shutdown_ptr)(char const *, plugin_shutdown_cb);
+// static int (*plugin_register_write_ptr)(char const *, plugin_write_cb,
+// user_data_t *);
+//
+// int meta_data_add_boolean_wrapper(meta_data_t *md, char const *key,
+// bool value) {
+// LOAD(meta_data_add_boolean);
+// return (*meta_data_add_boolean_ptr)(md, key, value);
+// }
+//
+// int meta_data_add_double_wrapper(meta_data_t *md, char const *key,
+// double value) {
+// LOAD(meta_data_add_double);
+// return (*meta_data_add_double_ptr)(md, key, value);
+// }
+//
+// int meta_data_add_signed_int_wrapper(meta_data_t *md, char const *key,
+// int64_t value) {
+// LOAD(meta_data_add_signed_int);
+// return (*meta_data_add_signed_int_ptr)(md, key, value);
+// }
+//
+// int meta_data_add_string_wrapper(meta_data_t *md, char const *key,
+// char const *value) {
+// LOAD(meta_data_add_string);
+// return (*meta_data_add_string_ptr)(md, key, value);
+// }
+//
+// int meta_data_add_unsigned_int_wrapper(meta_data_t *md, char const *key,
+// uint64_t value) {
+// LOAD(meta_data_add_unsigned_int);
+// return (*meta_data_add_unsigned_int_ptr)(md, key, value);
+// }
+//
+// meta_data_t *meta_data_create_wrapper(void) {
+// LOAD(meta_data_create);
+// return (*meta_data_create_ptr)();
+// }
+//
+// void meta_data_destroy_wrapper(meta_data_t *md) {
+// LOAD(meta_data_destroy);
+// (*meta_data_destroy_ptr)(md);
+// }
+//
+// int meta_data_get_boolean_wrapper(meta_data_t *md, char const *key,
+// bool *value) {
+// LOAD(meta_data_get_boolean);
+// return (*meta_data_get_boolean_ptr)(md, key, value);
+// }
+//
+// int meta_data_get_double_wrapper(meta_data_t *md, char const *key,
+// double *value) {
+// LOAD(meta_data_get_double);
+// return (*meta_data_get_double_ptr)(md, key, value);
+// }
+//
+// int meta_data_get_signed_int_wrapper(meta_data_t *md, char const *key,
+// int64_t *value) {
+// LOAD(meta_data_get_signed_int);
+// return (*meta_data_get_signed_int_ptr)(md, key, value);
+// }
+//
+// int meta_data_get_string_wrapper(meta_data_t *md, char const *key,
+// char **value) {
+// LOAD(meta_data_get_string);
+// return (*meta_data_get_string_ptr)(md, key, value);
+// }
+//
+// int meta_data_get_unsigned_int_wrapper(meta_data_t *md, char const *key,
+// uint64_t *value) {
+// LOAD(meta_data_get_unsigned_int);
+// return (*meta_data_get_unsigned_int_ptr)(md, key, value);
+// }
+//
+// int meta_data_toc_wrapper(meta_data_t *md, char ***toc) {
+// LOAD(meta_data_toc);
+// return (*meta_data_toc_ptr)(md, toc);
+// }
+//
+// int meta_data_type_wrapper(meta_data_t *md, char const *key) {
+// LOAD(meta_data_type);
+// return (*meta_data_type_ptr)(md, key);
+// }
+//
+// int plugin_dispatch_values_wrapper(value_list_t const *vl) {
+// LOAD(plugin_dispatch_values);
+// return (*plugin_dispatch_values_ptr)(vl);
+// }
+//
+// cdtime_t plugin_get_interval_wrapper(void) {
+// LOAD(plugin_get_interval);
+// return (*plugin_get_interval_ptr)();
+// }
+//
+// int plugin_register_complex_read_wrapper(meta_data_t *group, char const *name,
+// plugin_read_cb callback,
+// cdtime_t interval, user_data_t *ud) {
+// LOAD(plugin_register_complex_read);
+// return (*plugin_register_complex_read_ptr)(group, name, callback, interval,
+// ud);
+// }
+//
+// int plugin_register_log_wrapper(char const *name, plugin_log_cb callback,
+// user_data_t *ud) {
+// LOAD(plugin_register_log);
+// return (*plugin_register_log_ptr)(name, callback, ud);
+// }
+//
+// int plugin_register_shutdown_wrapper(char const *name,
+// plugin_shutdown_cb callback) {
+// LOAD(plugin_register_shutdown);
+// return (*plugin_register_shutdown_ptr)(name, callback);
+// }
+//
+// int plugin_register_write_wrapper(char const *name, plugin_write_cb callback,
+// user_data_t *ud) {
+// LOAD(plugin_register_write);
+// return (*plugin_register_write_ptr)(name, callback, ud);
+// }
+import "C"
diff --git a/rpc/client.go b/rpc/client.go
index a3d081b..824f366 100644
--- a/rpc/client.go
+++ b/rpc/client.go
@@ -15,7 +15,7 @@ type client struct {
pb.CollectdClient
}
-// Newclient returns a wrapper around the gRPC client connection that maps
+// NewClient returns a wrapper around the gRPC client connection that maps
// between the Go interface and the gRPC interface.
func NewClient(conn *grpc.ClientConn) Interface {
return &client{
Debdiff
[The following lists of changes regard files as different if they have different names, permissions or owners.]
Files in second set of .debs but not in first
-rw-r--r-- root/root /usr/share/gocode/src/collectd.org/config/config.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/config/config_test.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/exec/exec_x_test.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/format/putval_test.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/meta/meta.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/meta/meta_test.go -rw-r--r-- root/root /usr/share/gocode/src/collectd.org/network/network_x_test.go
No differences were encountered in the control files