Add starlark processor (influxdata#7660)

agent/agent.go

@@ -508,7 +508,9 @@ func (a *Agent) runProcessors(
 
 			acc := NewAccumulator(unit.processor, unit.dst)
 			for m := range unit.src {
-				unit.processor.Add(m, acc)
+				if err := unit.processor.Add(m, acc); err != nil {
+					acc.AddError(err)
+				}
 			}
 			unit.processor.Stop()
 			close(unit.dst)

docs/LICENSE_OF_DEPENDENCIES.md

@@ -55,6 +55,7 @@ following works:
 - github.com/go-ole/go-ole [MIT License](https://github.com/go-ole/go-ole/blob/master/LICENSE)
 - github.com/go-redis/redis [BSD 2-Clause "Simplified" License](https://github.com/go-redis/redis/blob/master/LICENSE)
 - github.com/go-sql-driver/mysql [Mozilla Public License 2.0](https://github.com/go-sql-driver/mysql/blob/master/LICENSE)
+- go.starlark.net [BSD 3-Clause "New" or "Revised" License](https://github.com/google/starlark-go/blob/master/LICENSE)
 - github.com/goburrow/modbus [BSD 3-Clause "New" or "Revised" License](https://github.com/goburrow/modbus/blob/master/LICENSE)
 - github.com/goburrow/serial [MIT License](https://github.com/goburrow/serial/LICENSE)
 - github.com/gobwas/glob [MIT License](https://github.com/gobwas/glob/blob/master/LICENSE)

go.mod

@@ -126,6 +126,7 @@ require (
 	github.com/wvanbergen/kafka v0.0.0-20171203153745-e2edea948ddf
 	github.com/wvanbergen/kazoo-go v0.0.0-20180202103751-f72d8611297a // indirect
 	github.com/yuin/gopher-lua v0.0.0-20180630135845-46796da1b0b4 // indirect
+	go.starlark.net v0.0.0-20191227232015-caa3e9aa5008
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6 // indirect
 	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
 	golang.org/x/net v0.0.0-20200301022130-244492dfa37a

go.sum

@@ -581,6 +581,8 @@ go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3 h1:8sGtKOrtQqkN1bp2AtX+misvLIlOmsEsNd+9NIcPEm8=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+go.starlark.net v0.0.0-20191227232015-caa3e9aa5008 h1:PUpdYMZifLwPlUnFfT/2Hkqr7p0SSpOR7xrDiPaD52k=
+go.starlark.net v0.0.0-20191227232015-caa3e9aa5008/go.mod h1:nmDLcffg48OtT/PSW0Hg7FvpRQsQh5OSqIylirxKC7o=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -701,6 +703,8 @@ golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456 h1:ng0gs1AKnRRuEMZoTLLlbOd+C17zUDepwGQBb/n+JVg=
 golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191002063906-3421d5a6bb1c h1:Vco5b+cuG5NNfORVxZy6bYZQ7rsigisU1WQFkvQ0L5E=
+golang.org/x/sys v0.0.0-20191002063906-3421d5a6bb1c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191003212358-c178f38b412c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

models/running_processor.go

@@ -78,21 +78,21 @@ func (r *RunningProcessor) Start(acc telegraf.Accumulator) error {
 	return r.Processor.Start(acc)
 }
 
-func (r *RunningProcessor) Add(m telegraf.Metric, acc telegraf.Accumulator) {
+func (r *RunningProcessor) Add(m telegraf.Metric, acc telegraf.Accumulator) error {
 	if ok := r.Config.Filter.Select(m); !ok {
 		// pass downstream
 		acc.AddMetric(m)
-		return
+		return nil
 	}
 
 	r.Config.Filter.Modify(m)
 	if len(m.FieldList()) == 0 {
 		// drop metric
 		r.metricFiltered(m)
-		return
+		return nil
 	}
 
-	r.Processor.Add(m, acc)
+	return r.Processor.Add(m, acc)
 }
 
 func (r *RunningProcessor) Stop() {

plugins/processors/all/all.go

@@ -16,6 +16,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/processors/regex"
 	_ "github.com/influxdata/telegraf/plugins/processors/rename"
 	_ "github.com/influxdata/telegraf/plugins/processors/s2geo"
+	_ "github.com/influxdata/telegraf/plugins/processors/starlark"
 	_ "github.com/influxdata/telegraf/plugins/processors/strings"
 	_ "github.com/influxdata/telegraf/plugins/processors/tag_limit"
 	_ "github.com/influxdata/telegraf/plugins/processors/template"

plugins/processors/starlark/README.md

@@ -0,0 +1,150 @@
+# Starlark Processor
+
+The `starlark` processor calls a Starlark function for each matched metric,
+allowing for custom programmatic metric processing.
+
+The Starlark language is a dialect of Python, and will be familiar to those who
+have experience with the Python language.  However, keep in mind that it is not
+Python and that there are major syntax [differences][#Python Differences].
+Existing Python code is unlikely to work unmodified.  The execution environment
+is sandboxed, and it is not possible to do I/O operations such as reading from
+files or sockets.
+
+The Starlark [specification][] has details about the syntax and available
+functions.
+
+### Configuration
+
+```toml
+[[processors.starlark]]
+  ## The Starlark source can be set as a string in this configuration file, or
+  ## by referencing a file containing the script.  Only one source or script
+  ## should be set at once.
+  ##
+  ## Source of the Starlark script.
+  source = '''
+def apply(metric):
+	return metric
+'''
+
+  ## File containing a Starlark script.
+  # script = "/usr/local/bin/myscript.star"
+```
+
+### Usage
+
+The script should contain a function called `apply` that takes the metric as
+its single argument.  The function will be called with each metric, and can
+return `None`, a single metric, or a list of metrics.
+```python
+def apply(metric):
+	return metric
+```
+
+Reference the Starlark [specification][] to see the list of available types and
+functions that can be used in the script.  In addition to these the following
+types and functions are exposed to the script.
+
+**Metric(*name*)**:
+Create a new metric with the given measurement name.  The metric will have no
+tags or fields and defaults to the current time.
+
+- **name**:
+The name is a [string][string] containing the metric measurement name.
+
+- **tags**:
+A [dict-like][dict] object containing the metric's tags.
+
+
+- **fields**:
+A [dict-like][dict] object containing the metric's fields.  The values may be
+of type int, float, string, or bool.
+
+- **time**:
+The timestamp of the metric as an integer in nanoseconds since the Unix
+epoch.
+
+**deepcopy(*metric*)**: Make a copy of an existing metric.
+
+### Python Differences
+
+While Starlark is similar to Python it is not the same.
+
+- Starlark has limited support for error handling and no exceptions.  If an
+  error occurs the script will immediately end and Telegraf will drop the
+  metric.  Check the Telegraf logfile for details about the error.
+
+- It is not possible to import other packages and the Python standard library
+  is not available.  As such, it is not possible to open files or sockets.
+
+- These common keywords are **not supported** in the Starlark grammar:
+  ```
+  as             finally        nonlocal
+  assert         from           raise
+  class          global         try
+  del            import         with
+  except         is             yield
+  ```
+
+### Common Questions
+
+**How can I drop/delete a metric?**
+
+If you don't return the metric it will be deleted.  Usually this means the
+function should `return None`.
+
+**How should I make a copy of a metric?**
+
+Use `deepcopy(metric)` to create a copy of the metric.
+
+**How can I return multiple metrics?**
+
+You can return a list of metrics:
+```python
+def apply(metric):
+    m2 = deepcopy(metric)
+    return [metric, m2]
+```
+
+**What happens to a tracking metric if an error occurs in the script?**
+
+The metric is marked as undelivered.
+
+**How do I create a new metric?**
+
+Use the `Metric(name)` function and set at least one field.
+
+**What is the fastest way to iterate over tags/fields?**
+
+The fastest way to iterate is to use a for-loop on the tags or fields attribute:
+```python
+def apply(metric):
+    for k in metric.tags:
+        pass
+    return metric
+```
+
+When you use this form, it is not possible to modify the tags inside the loop,
+if this is needed you should use the `.keys()`, `.values()`, or `.items()` forms:
+```python
+def apply(metric):
+    for k, v in metric.tags.items():
+        pass
+    return metric
+```
+
+**How can I save values across multiple calls to the script?**
+
+Telegraf freezes the global scope, which prevents it from being modified.
+Attempting to modify the global scope will fail with an error.
+
+
+### Examples
+
+- [ratio](/plugins/processors/starlark/testdata/ratio.star)
+- [rename](/plugins/processors/starlark/testdata/rename.star)
+- [scale](/plugins/processors/starlark/testdata/scale.star)
+
+[specification]: https://github.com/google/starlark-go/blob/master/doc/spec.md
+[string]: https://github.com/google/starlark-go/blob/master/doc/spec.md#strings
+[dict]: https://github.com/google/starlark-go/blob/master/doc/spec.md#dictionaries