Repository: influxdata/kapacitor
Pull request #2575 (2021-06-10 21:14:44)

feat(alerta-service): Support the "attributes" attribute in Alerta node
Fix influxdata#2349

This commit adds support for custom Alerta alert attributes (see Alerta Alert Format for details).

Attributes are specified with "attribute" keyword in tick script, e.g.:

    |alert()
        // ...
        .alerta()
            // ...
            .attribute('customUrl', 'https://example.com/')
            .attribute('booleanAttribute', TRUE)
            .attribute('numericAttribute', 9001)
            .attribute('templatedAttribute', 'Templates are supported: {{ .Name }}')

###### Required for all non-trivial PRs
- [x] Rebased/mergable
- [x] Tests pass
- [x] CHANGELOG.md updated
- [x] Sign [CLA](https://influxdata.com/community/cla/) (if not already signed)


sp1r: Tested this changes with Alerta API 8.6.0 (latest)
sp1r: Looks like builds on master branch are broken.

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 7b999b50..97a5c7ee 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -2,6 +2,13 @@
 version:  "2.1"
 
 executors:
+  cross-builder:
+    docker:
+      # NOTE: To upgrade the Go version, first push the upgrade to the cross-builder Dockerfile in the edge repo,
+      # then update the version here to match. Until we finish the migration to using the cross-builder image,
+      # you'll also need to update references to `cimg/go` and `GO_VERSION` in this file.
+      - image: quay.io/influxdb/cross-builder:go1.17.1-41d944d749b2c130119eff7efaaa9883ccffe5a7
+    resource_class: large
   linux-amd64:
     machine:
       image: ubuntu-2004:202107-02
@@ -14,9 +21,83 @@ commands:
       This will build and publish release packages for tag "$CIRCLE_TAG"
     steps:
       - run:
-          name: Deploy Release Packages
-          command:  |
-            ./build.sh --debug --clean --generate --package --package-udfs --upload --bucket=dl.influxdata.com/kapacitor/releases --platform=all --arch=all --release
+          name: Install additional dependencies
+          command: |
+            set -x
+            # APT packages
+            apt-get -qq update && apt-get -qq install -y \
+                software-properties-common \
+                unzip \
+                mercurial \
+                make \
+                ruby \
+                ruby-dev \
+                rpm \
+                zip \
+                python \
+                python-setuptools \
+                python3 \
+                python3-setuptools \
+                python3-boto \
+                autoconf \
+                automake \
+                libtool
+
+            # Ruby dependencies
+            gem install fpm
+
+            # Protobuf
+            wget -q https://github.com/google/protobuf/releases/download/v${PROTO_VERSION}/protobuf-python-${PROTO_VERSION}.tar.gz
+            tar -xf protobuf-python-${PROTO_VERSION}.tar.gz --no-same-owner
+            cd protobuf-${PROTO_VERSION}/python
+            python3 setup.py install
+            cd ../../
+            rm -rf /protobuf-${PROTO_VERSION} protobuf-python-${PROTO_VERSION}.tar.gz
+      - run:
+          name: Import GPG key
+          command: |
+            echo -e "$GPG_1X_PRIVATE_KEY" | gpg --batch --import --
+      - when:
+          condition:
+            and:
+              - << pipeline.git.tag >>
+              - not:
+                  matches:
+                    pattern: "(.*rc[0-9]+)|(.*test)$"
+                    value: << pipeline.git.tag >>
+          steps:
+            - run:
+                name: Import GPG key
+                command: |
+                  echo -e "$GPG_1X_PRIVATE_KEY" | gpg --import --
+            - run:
+                name: Deploy Release Packages
+                command:  |
+                  ./build.py --debug --clean --generate --package --package-udfs --upload --bucket=dl.influxdata.com/kapacitor/releases --platform=all --arch=all --release --checksum --sign
+      - when:
+          condition:
+            or:
+              - not: << pipeline.git.tag >>
+              - matches:
+                  pattern: ".*rc[0-9]+$"
+                  value: << pipeline.git.tag >>
+          steps:
+            - run:
+                name: Deploy Release Packages
+                command:  |
+                  ./build.py --debug --clean --generate --package --package-udfs --upload --bucket=dl.influxdata.com/kapacitor/releases --platform=all --arch=all --release --checksum
+      - when:
+          condition:
+            matches:
+              pattern: "^.*test$"
+              value: << pipeline.git.tag >>
+          steps:
+            - run:
+                name: Deploy Non-Release Packages
+                command:  |
+                  ./build.py --debug --clean --generate --package --package-udfs --platform=all --arch=all --checksum --sign
+      - store_artifacts:
+          path: ./build
 
   run_tests:
     description:  >
@@ -34,7 +115,7 @@ commands:
       - run:
           name: Deploy Nightly Build
           command:  |
-            ./build.sh --debug --clean --generate --package --package-udfs --upload --bucket=dl.influxdata.com/kapacitor/releases/nightly --platform=all --arch=all --nightly
+            ./build.py --debug --clean --generate --package --package-udfs --upload --bucket=dl.influxdata.com/kapacitor/releases/nightly --platform=all --arch=all --nightly
 
 jobs:
   build:
@@ -46,14 +127,14 @@ jobs:
       - run_tests
 
   release:
-    executor: linux-amd64
+    executor: cross-builder
     working_directory: ~/kapacitor
     steps:
       - checkout
       - deploy_release_packages
 
   nightly-build:
-    executor: linux-amd64
+    executor: cross-builder
     working_directory: ~/kapacitor
     steps:
       - checkout
@@ -74,7 +155,9 @@ workflows:
             - build
           filters:
             tags:
-              only:  /^v[0-9]+(\.[0-9]+){2}(-(rc|beta)[0-9]+)?/
+              only:
+                - /^v[0-9]+(\.[0-9]+){2}(-(rc|beta)[0-9]+)?/
+                - /^.*test$/
             branches:
               ignore:  /.*/
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca673e10..c4fa303b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,6 @@
 # Changelog
 
-## v1.6.2 [2021-08-04]
+## v1.6.2 [2021-09-24]
 
 ### Bugfixes
 - [#2582](https://github.com/influxdata/kapacitor/pull/2582): expvar string json encoding to handle special chars, thanks @prashanthjbabu! 
@@ -11,6 +11,8 @@
 ### Features
 
 - [#2621](https://github.com/influxdata/kapacitor/pull/2621): Add template-id to task list.
+- [#2622](https://github.com/influxdata/kapacitor/pull/2622): auto-create 1.x DB or 2.x bucket for flux task logs
+- [#2608](https://github.com/influxdata/kapacitor/pull/2608): Allow for compact json in templates and BP
 
 ## v1.6.1 [2021-07-22]
 
@@ -35,6 +37,7 @@
 - [#2559](https://github.com/influxdata/kapacitor/pull/2559): kapacitor cli supports flux tasks
 - [#2560](https://github.com/influxdata/kapacitor/pull/2560): enable new-style slack apps
 - [#2562](https://github.com/influxdata/kapacitor/pull/2560): handle Delete messages in the joinNode
+- [#2575](https://github.com/influxdata/kapacitor/pull/2575): Support the "attributes" attribute in Alerta node
 - [#2576](https://github.com/influxdata/kapacitor/pull/2576): shared secret auth to influxdb in OSS
 
 ### Bugfixes
diff --git a/Dockerfile_build_ubuntu64 b/Dockerfile_build_ubuntu64
index d0fe6aff..33b7f162 100644
--- a/Dockerfile_build_ubuntu64
+++ b/Dockerfile_build_ubuntu64
@@ -1,5 +1,5 @@
 ARG GO_VERSION
-FROM quay.io/influxdb/cross-builder:go${GO_VERSION}-23321b2fc172da7a9730893d885056353c9b11ac
+FROM quay.io/influxdb/cross-builder:go${GO_VERSION}-41d944d749b2c130119eff7efaaa9883ccffe5a7
 
 # This dockerfile is capabable of performing all
 # build/test/package/deploy actions needed for Kapacitor.
diff --git a/alert.go b/alert.go
index 82513381..c8632dcd 100644
--- a/alert.go
+++ b/alert.go
@@ -86,6 +86,7 @@ type AlertNode struct {
 
 // Create a new  AlertNode which caches the most recent item and exposes it over the HTTP API.
 func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (an *AlertNode, err error) {
+	const oneMeg = 2 << 19
 	ctx := []keyvalue.T{
 		keyvalue.KV("task", et.Task.ID),
 	}
@@ -119,16 +120,36 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 	}
 
 	an.detailsTmpl, err = html.New("details").Funcs(html.FuncMap{
-		"json": func(v interface{}) html.JS {
-
+		"jsonCompact": func(v interface{}) html.JS {
 			tmpBuffer := an.bufPool.Get().(*bytes.Buffer)
+			tmpBuffer2 := an.bufPool.Get().(*bytes.Buffer)
+
 			defer func() {
-				tmpBuffer.Reset()
-				an.bufPool.Put(tmpBuffer)
+				if tmpBuffer.Cap() < oneMeg { // only reuse the buffer if it is less than 500kb
+					tmpBuffer.Reset()
+					an.bufPool.Put(tmpBuffer)
+				}
+				if tmpBuffer2.Cap() < oneMeg { // only reuse the buffer if it is less than 500kb
+					tmpBuffer2.Reset()
+					an.bufPool.Put(tmpBuffer2)
+				}
 			}()
 
 			_ = json.NewEncoder(tmpBuffer).Encode(v)
+			_ = json.Compact(tmpBuffer2, tmpBuffer.Bytes())
+			return html.JS(tmpBuffer2.String())
+		},
+		"json": func(v interface{}) html.JS {
+			tmpBuffer := an.bufPool.Get().(*bytes.Buffer)
 
+			defer func() {
+				if tmpBuffer.Cap() < oneMeg { // only reuse the buffer if it is less than 500kb
+					tmpBuffer.Reset()
+					an.bufPool.Put(tmpBuffer)
+				}
+			}()
+
+			_ = json.NewEncoder(tmpBuffer).Encode(v)
 			return html.JS(tmpBuffer.String())
 		},
 	}).Parse(n.Details)
@@ -381,6 +402,9 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 		if len(a.Correlate) != 0 {
 			c.Correlate = a.Correlate
 		}
+		if len(a.Attributes) != 0 {
+			c.Attributes = a.Attributes
+		}
 		if a.Timeout != 0 {
 			c.Timeout = a.Timeout
 		}
diff --git a/barrier.go b/barrier.go
index d1f122ad..da3bd093 100644
--- a/barrier.go
+++ b/barrier.go
@@ -2,10 +2,9 @@ package kapacitor
 
 import (
 	"errors"
-	"time"
-
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/influxdata/kapacitor/edge"
 	"github.com/influxdata/kapacitor/models"
diff --git a/build.py b/build.py
index 4b5b8cda..9144cdfe 100755
--- a/build.py
+++ b/build.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 import argparse
 import hashlib
@@ -879,6 +879,17 @@ def main(args):
         if args.package_udfs:
             packages += package_udfs(args.version, args.outdir)
 
+        if args.checksum:
+            logging.debug("Generating checksums for packages: {}".format(packages))
+            checksums = []
+            for p in packages:
+                checksum = generate_sha256_from_file(p)
+                sha_filename = p + '.sha256'
+                with open(sha_filename, 'w+') as fp:
+                    fp.write(checksum)
+                checksums.append(sha_filename)
+
+            packages += checksums
         if args.sign:
             logging.debug("Generating GPG signatures for packages: {}".format(packages))
             sigs = [] # retain signatures so they can be uploaded with packages
@@ -895,10 +906,9 @@ def main(args):
                 args.upload_overwrite = True
             if not upload_packages(packages, bucket_name=args.bucket, overwrite=args.upload_overwrite):
                 return 1
-        logging.info("Packages created:")
+        logging.info("Packages and files created:")
         for p in packages:
-            logging.info("{} (sha256={})".format(p.split('/')[-1:][0],
-                                              generate_sha256_from_file(p)))
+            logging.info("{}".format(p.split('/')[-1:][0]))
 
 
     if orig_branch != get_current_branch():
@@ -1017,6 +1027,9 @@ if __name__ == '__main__':
     parser.add_argument('--sign',
                         action='store_true',
                         help='Create GPG detached signatures for packages (when package is specified)')
+    parser.add_argument('--checksum',
+                        action='store_true',
+                        help='Create md5 and sha256 checksums for packages (when package is specified)')
     parser.add_argument('--test',
                         action='store_true',
                         help='Run tests (does not produce build output)')
diff --git a/build.sh b/build.sh
index d832b198..1f453f46 100755
--- a/build.sh
+++ b/build.sh
@@ -12,7 +12,7 @@ BUILD_NUM=${BUILD_NUM-$RANDOM}
 # Home dir of the docker user
 HOME_DIR=/root
 
-GO_VERSION=1.17
+GO_VERSION=1.17.1
 
 imagename=kapacitor-builder-img-$BUILD_NUM
 
diff --git a/http_post.go b/http_post.go
index e774ef8b..58136f19 100644
--- a/http_post.go
+++ b/http_post.go
@@ -1,6 +1,8 @@
 package kapacitor
 
 import (
+	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -9,9 +11,6 @@ import (
 	"sync"
 	"time"
 
-	"bytes"
-	"context"
-
 	"github.com/influxdata/kapacitor/edge"
 	"github.com/influxdata/kapacitor/keyvalue"
 	"github.com/influxdata/kapacitor/models"
diff --git a/influxdb/client.go b/influxdb/client.go
index 108de1d3..992b72de 100644
--- a/influxdb/client.go
+++ b/influxdb/client.go
@@ -52,6 +52,10 @@ type Client interface {
 	// if it exists.  Unlike QueryFlux, this returns a *Response
 	// object.
 	QueryFluxResponse(q FluxQuery) (*Response, error)
+
+	// CreateBucketV2 uses the 2.x /api/v2/bucket api to create a bucket.
+	// Note that 1.x does not support this API
+	CreateBucketV2(bucket string, org string, orgID string) error
 }
 
 type ClientUpdater interface {
@@ -147,6 +151,69 @@ type HTTPClient struct {
 	compression string
 }
 
+func (c *HTTPClient) getOrgID(org string) (string, error) {
+	u := c.url()
+	u.Path = "/api/v2/orgs"
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return "", err
+	}
+	reader, err := c.doHttpV2(req, 200)
+	if err != nil {
+		return "", err
+	}
+	var val struct {
+		Orgs []struct {
+			ID   string `json:"id"`
+			Name string `json:"name"`
+		} `json:"orgs"`
+	}
+	err = json.NewDecoder(reader).Decode(&val)
+	if err != nil {
+		return "", fmt.Errorf("decoding json from org request: %w", err)
+	}
+	// Check for "org" matching id to account for odd flux query API that will accept an org ID as an org name
+	for _, o := range val.Orgs {
+		if o.ID == org {
+			return o.ID, nil
+		}
+	}
+	for _, o := range val.Orgs {
+		if o.Name == org {
+			return o.ID, nil
+		}
+	}
+	return "", fmt.Errorf("unknown organization name %s", org)
+}
+
+func (c *HTTPClient) CreateBucketV2(bucket, org, orgID string) error {
+	u := c.url()
+	u.Path = "/api/v2/buckets"
+
+	var err error
+	if orgID == "" {
+		orgID, err = c.getOrgID(org)
+		if err != nil {
+			return fmt.Errorf("attempting to get org id for org: %w", err)
+		}
+	}
+
+	body := fmt.Sprintf(`{"orgID": %q, "name": %q, "retentionRules": [] }`, orgID, bucket)
+
+	req, err := http.NewRequest("POST", u.String(), bytes.NewBufferString(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept-Encoding", "gzip")
+	reader, err := c.doHttpV2(req, 201)
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(io.Discard, reader)
+	return err
+}
+
 // NewHTTPClient returns a new Client from the provided config.
 // Client is safe for concurrent use by multiple goroutines.
 func NewHTTPClient(conf Config) (*HTTPClient, error) {
@@ -367,7 +434,7 @@ func (r readClose) Close() error {
 
 var _ io.ReadCloser = readClose{}
 
-func (c *HTTPClient) doFlux(req *http.Request, codes ...int) (io.ReadCloser, error) {
+func (c *HTTPClient) doHttpV2(req *http.Request, codes ...int) (io.ReadCloser, error) {
 	// Get current config
 	config := c.loadConfig()
 	// Set auth credentials
@@ -422,7 +489,7 @@ func (c *HTTPClient) doFlux(req *http.Request, codes ...int) (io.ReadCloser, err
 			Error   string `json:"error"`
 		}{}
 		if err := d.Decode(&rp); err != nil {
-			return nil, err
+			return nil, fmt.Errorf("error attempting to decode http response, status %d: %w", resp.StatusCode, err)
 		}
 		if rp.Error != "" {
 			return nil, errors.New(rp.Error)
@@ -593,9 +660,7 @@ type Result struct {
 
 func (c *HTTPClient) buildFluxRequest(q FluxQuery) (*http.Request, error) {
 	u := c.url()
-	if c.url().Path == "" || c.url().Path == "/" {
-		u.Path = "/api/v2/query"
-	}
+	u.Path = "/api/v2/query"
 
 	v := url.Values{}
 	if q.Org != "" {
@@ -645,7 +710,7 @@ func (c *HTTPClient) QueryFluxResponse(q FluxQuery) (*Response, error) {
 	if err != nil {
 		return nil, err
 	}
-	reader, err := c.doFlux(req, http.StatusOK)
+	reader, err := c.doHttpV2(req, http.StatusOK)
 	if err != nil {
 		return nil, err
 	}
@@ -663,7 +728,7 @@ func (c *HTTPClient) QueryFlux(q FluxQuery) (flux.ResultIterator, error) {
 	if err != nil {
 		return nil, err
 	}
-	reader, err := c.doFlux(req, http.StatusOK)
+	reader, err := c.doHttpV2(req, http.StatusOK)
 	if err != nil {
 		return nil, err
 	}
diff --git a/influxdb/token_client.go b/influxdb/token_client.go
index d18aa8ab..6f25a829 100644
--- a/influxdb/token_client.go
+++ b/influxdb/token_client.go
@@ -121,6 +121,10 @@ func (tc *tokenClient) QueryFluxResponse(q FluxQuery) (*Response, error) {
 	return tc.client.QueryFluxResponse(q)
 }
 
+func (tc *tokenClient) CreateBucketV2(bucket, org, orgID string) error {
+	return tc.client.CreateBucketV2(bucket, org, orgID)
+}
+
 func (tc *tokenClient) Open() error {
 	tc.mu.Lock()
 	defer tc.mu.Unlock()
diff --git a/pipeline/alert.go b/pipeline/alert.go
index ca21ea50..1a71ac7f 100644
--- a/pipeline/alert.go
+++ b/pipeline/alert.go
@@ -1263,6 +1263,9 @@ type AlertaHandler struct {
 	// List of Correlated
 	Correlate []string `tick:"Correlated" json:"correlate"`
 
+	// Map of alert Attributes
+	Attributes map[string]interface{} `tick:"Attribute" json:"attributes"`
+
 	// Alerta timeout.
 	// Default: 24h
 	Timeout time.Duration `json:"timeout"`
@@ -1281,6 +1284,14 @@ func (a *AlertaHandler) Correlated(correlate ...string) *AlertaHandler {
 	return a
 }
 
+func (a *AlertaHandler) Attribute(k string, v interface{}) *AlertaHandler {
+	if a.Attributes == nil {
+		a.Attributes = make(map[string]interface{})
+	}
+	a.Attributes[k] = v
+	return a
+}
+
 // Send alert to an MQTT broker
 // tick:property
 func (n *AlertNodeData) Mqtt(topic string) *MQTTHandler {
diff --git a/pipeline/tick/alert.go b/pipeline/tick/alert.go
index b80eb41c..670d66a9 100644
--- a/pipeline/tick/alert.go
+++ b/pipeline/tick/alert.go
@@ -227,6 +227,15 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 			Dot("services", args(h.Service)...).
 			Dot("correlated", args(h.Correlate)...).
 			Dot("timeout", h.Timeout)
+
+		var attributes []string
+		for k := range h.Attributes {
+			attributes = append(attributes, k)
+		}
+		sort.Strings(attributes)
+		for _, k := range attributes {
+			n.Dot("attribute", k, h.Attributes[k])
+		}
 	}
 
 	for _, h := range a.OpsGenieHandlers {
diff --git a/services/alert/dao_easyjson.go b/services/alert/dao_easyjson.go
index e4399555..9b13b26f 100644
--- a/services/alert/dao_easyjson.go
+++ b/services/alert/dao_easyjson.go
@@ -4,11 +4,10 @@ package alert
 
 import (
 	json "encoding/json"
-	time "time"
-
 	easyjson "github.com/mailru/easyjson"
 	jlexer "github.com/mailru/easyjson/jlexer"
 	jwriter "github.com/mailru/easyjson/jwriter"
+	time "time"
 )
 
 // suppress unused package warning
diff --git a/services/alerta/service.go b/services/alerta/service.go
index d82f5cd1..9391bbc3 100644
--- a/services/alerta/service.go
+++ b/services/alerta/service.go
@@ -51,17 +51,18 @@ func NewService(c Config, d Diagnostic) *Service {
 }
 
 type testOptions struct {
-	Resource    string   `json:"resource"`
-	Event       string   `json:"event"`
-	Environment string   `json:"environment"`
-	Severity    string   `json:"severity"`
-	Group       string   `json:"group"`
-	Value       string   `json:"value"`
-	Message     string   `json:"message"`
-	Origin      string   `json:"origin"`
-	Service     []string `json:"service"`
-	Correlate   []string `json:"correlate"`
-	Timeout     string   `json:"timeout"`
+	Resource    string                 `json:"resource"`
+	Event       string                 `json:"event"`
+	Environment string                 `json:"environment"`
+	Severity    string                 `json:"severity"`
+	Group       string                 `json:"group"`
+	Value       string                 `json:"value"`
+	Message     string                 `json:"message"`
+	Origin      string                 `json:"origin"`
+	Service     []string               `json:"service"`
+	Correlate   []string               `json:"correlate"`
+	Attributes  map[string]interface{} `json:"attributes"`
+	Timeout     string                 `json:"timeout"`
 }
 
 func (s *Service) TestOptions() interface{} {
@@ -77,7 +78,12 @@ func (s *Service) TestOptions() interface{} {
 		Origin:      c.Origin,
 		Service:     []string{"testServiceA", "testServiceB"},
 		Correlate:   []string{"testServiceX", "testServiceY"},
-		Timeout:     "24h0m0s",
+		Attributes: map[string]interface{}{
+			"testAttributeA": "A",
+			"testAttributeB": true,
+			"testAttributeC": 9001.0,
+		},
+		Timeout: "24h0m0s",
 	}
 }
 
@@ -101,6 +107,7 @@ func (s *Service) Test(options interface{}) error {
 		o.Origin,
 		o.Service,
 		o.Correlate,
+		o.Attributes,
 		timeout,
 		map[string]string{},
 		models.Result{},
@@ -133,12 +140,12 @@ func (s *Service) Update(newConfig []interface{}) error {
 	return nil
 }
 
-func (s *Service) Alert(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin string, service []string, correlate []string, timeout time.Duration, tags map[string]string, data models.Result) error {
+func (s *Service) Alert(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin string, service []string, correlate []string, attributes map[string]interface{}, timeout time.Duration, tags map[string]string, data models.Result) error {
 	if resource == "" || event == "" {
 		return errors.New("Resource and Event are required to send an alert")
 	}
 
-	req, err := s.preparePost(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin, service, correlate, timeout, tags, data)
+	req, err := s.preparePost(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin, service, correlate, attributes, timeout, tags, data)
 	if err != nil {
 		return err
 	}
@@ -166,7 +173,7 @@ func (s *Service) Alert(token, tokenPrefix, resource, event, environment, severi
 	return nil
 }
 
-func (s *Service) preparePost(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin string, service []string, correlate []string, timeout time.Duration, tags map[string]string, data models.Result) (*http.Request, error) {
+func (s *Service) preparePost(token, tokenPrefix, resource, event, environment, severity, group, value, message, origin string, service []string, correlate []string, attributes map[string]interface{}, timeout time.Duration, tags map[string]string, data models.Result) (*http.Request, error) {
 	c := s.config()
 
 	if !c.Enabled {
@@ -215,6 +222,9 @@ func (s *Service) preparePost(token, tokenPrefix, resource, event, environment,
 	if len(correlate) > 0 {
 		postData["correlate"] = correlate
 	}
+	if len(attributes) > 0 {
+		postData["attributes"] = attributes
+	}
 	postData["timeout"] = int64(timeout / time.Second)
 
 	tagList := make([]string, 0)
@@ -284,6 +294,9 @@ type HandlerConfig struct {
 	// List of correlated events
 	Correlate []string `mapstructure:"correlate"`
 
+	// Map of alert attributes
+	Attributes map[string]interface{} `mapstructure:"attributes"`
+
 	// Alerta timeout.
 	// Default: 24h
 	Timeout time.Duration `mapstructure:"timeout"`
@@ -301,6 +314,8 @@ type handler struct {
 	groupTmpl       *text.Template
 	serviceTmpl     []*text.Template
 	correlateTmpl   []*text.Template
+
+	attributes map[string]interface{}
 }
 
 func (s *Service) DefaultHandlerConfig() HandlerConfig {
@@ -353,6 +368,21 @@ func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) (alert.Handler, er
 		ctmpl = append(ctmpl, tmpl)
 	}
 
+	attrs := make(map[string]interface{})
+	for k, v := range c.Attributes {
+		switch value := v.(type) {
+		case string:
+			// resolve templates
+			tmpl, err := text.New(k).Parse(value)
+			if err != nil {
+				return nil, err
+			}
+			attrs[k] = tmpl
+		default:
+			attrs[k] = value
+		}
+	}
+
 	return &handler{
 		s:               s,
 		c:               c,
@@ -364,6 +394,7 @@ func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) (alert.Handler, er
 		valueTmpl:       vtmpl,
 		serviceTmpl:     stmpl,
 		correlateTmpl:   ctmpl,
+		attributes:      attrs,
 	}, nil
 }
 
@@ -464,6 +495,24 @@ func (h *handler) Handle(event alert.Event) {
 		}
 	}
 
+	attributes := make(map[string]interface{})
+	if len(h.attributes) != 0 {
+		for k, v := range h.attributes {
+			switch value := v.(type) {
+			case *text.Template:
+				err = value.Execute(&buf, td)
+				if err != nil {
+					h.diag.TemplateError(err, keyvalue.KV("attributes", value.Name()))
+					return
+				}
+				attributes[k] = buf.String()
+				buf.Reset()
+			default:
+				attributes[k] = value
+			}
+		}
+	}
+
 	var severity string
 
 	switch event.State.Level {
@@ -492,6 +541,7 @@ func (h *handler) Handle(event alert.Event) {
 		h.c.Origin,
 		service,
 		correlate,
+		attributes,
 		h.c.Timeout,
 		event.Data.Tags,
 		event.Data.Result,
diff --git a/services/auth/service.go b/services/auth/service.go
index 12eccf01..0dc333b4 100644
--- a/services/auth/service.go
+++ b/services/auth/service.go
@@ -22,7 +22,6 @@ import (
 	"github.com/influxdata/kapacitor/services/storage"
 	"github.com/influxdata/kapacitor/tlsconfig"
 	"github.com/pkg/errors"
-
 	"golang.org/x/crypto/bcrypt"
 )
 
diff --git a/services/bigpanda/service.go b/services/bigpanda/service.go
index 2f72a98e..7a641938 100644
--- a/services/bigpanda/service.go
+++ b/services/bigpanda/service.go
@@ -246,11 +246,15 @@ func (s *Service) preparePost(id string, message string, details string, level a
 		}
 	}
 
-	var post bytes.Buffer
-	enc := json.NewEncoder(&post)
+	var postTemp bytes.Buffer
+	enc := json.NewEncoder(&postTemp)
 	if err := enc.Encode(bpData); err != nil {
 		return nil, err
 	}
+	var post bytes.Buffer
+	if err := json.Compact(&post, postTemp.Bytes()); err != nil {
+		return nil, err
+	}
 
 	bpUrl := hc.URL
 	if bpUrl == "" {
diff --git a/services/discord/service.go b/services/discord/service.go
index 83a0b4b0..239b6081 100644
--- a/services/discord/service.go
+++ b/services/discord/service.go
@@ -7,11 +7,10 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sync"
 	text "text/template"
 	"time"
 
-	"sync"
-
 	"github.com/influxdata/kapacitor/alert"
 	khttp "github.com/influxdata/kapacitor/http"
 	"github.com/influxdata/kapacitor/keyvalue"
diff --git a/services/fluxtask/service.go b/services/fluxtask/service.go
index fb93d2f7..8cbb3e8b 100644
--- a/services/fluxtask/service.go
+++ b/services/fluxtask/service.go
@@ -2,6 +2,7 @@ package fluxtask
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/influxdata/influxdb/v2/kit/platform"
@@ -42,9 +43,16 @@ func (s *Service) Open() error {
 	if s.config.Enabled {
 		// create the task stack
 		s.kvService = kv.New(s.StorageService)
-		s.kvService.Open()
+		if err := s.kvService.Open(); err != nil {
+			return err
+		}
+		bucket := s.config.TaskRunBucket
+		if bucket == "" {
+			// Deal with previous default of empty string
+			bucket = task.DefaultTaskRunBucket
+		}
 		dataDestination := backend.DataDestination{
-			Bucket:      s.config.TaskRunBucket,
+			Bucket:      bucket,
 			Org:         s.config.TaskRunOrg,
 			OrgID:       s.config.TaskRunOrgID,
 			Measurement: s.config.TaskRunMeasurement,
@@ -60,13 +68,16 @@ func (s *Service) Open() error {
 			if err != nil {
 				return err
 			}
-			combinedTaskService := backend.NewAnalyticalStorage(
+			combinedTaskService, err := backend.NewAnalyticalStorage(
 				s.logger.With(zap.String("service", "fluxtask-analytical-store")),
 				taskService,
 				taskControlService,
 				cli,
 				dataDestination,
 			)
+			if err != nil {
+				return fmt.Errorf("creating analytical store: %w", err)
+			}
 			taskService = combinedTaskService
 			taskControlService = combinedTaskService
 		}
diff --git a/services/httppost/config.go b/services/httppost/config.go
index 4eb52b5b..8f6858fc 100644
--- a/services/httppost/config.go
+++ b/services/httppost/config.go
@@ -111,6 +111,13 @@ func GetTemplate(tmpl, tpath string) (*template.Template, error) {
 				_ = json.NewEncoder(&buf).Encode(v)
 				return buf.String()
 			},
+			"jsonCompact": func(v interface{}) string {
+				tmpBuffer := bytes.Buffer{}
+				tmpBuffer2 := bytes.Buffer{}
+				_ = json.NewEncoder(&tmpBuffer).Encode(v)
+				_ = json.Compact(&tmpBuffer2, tmpBuffer.Bytes())
+				return tmpBuffer2.String()
+			},
 		}).Parse(tmpl)
 		return t, errors.Wrap(err, "failed to parse template")
 	}
diff --git a/services/httppost/service.go b/services/httppost/service.go
index 0ae1b660..eadce638 100644
--- a/services/httppost/service.go
+++ b/services/httppost/service.go
@@ -2,6 +2,7 @@ package httppost
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
@@ -12,11 +13,8 @@ import (
 	"strings"
 	"sync"
 	"text/template"
-
 	"time"
 
-	"context"
-
 	"github.com/influxdata/kapacitor/alert"
 	khttp "github.com/influxdata/kapacitor/http"
 	"github.com/influxdata/kapacitor/keyvalue"
diff --git a/services/influxdb/service.go b/services/influxdb/service.go
index 54746e95..dc45d95c 100644
--- a/services/influxdb/service.go
+++ b/services/influxdb/service.go
@@ -563,6 +563,11 @@ func (c *influxdbCluster) Open() error {
 
 	c.watchSubs()
 
+	// Even if we're not linking subscriptions, validate the client to ensure influxdb is actually up
+	if err := c.validateClientWithBackoff(ctx); err != nil {
+		return errors.Wrap(err, "failed to validate influxdb client on startup")
+	}
+
 	if err := c.linkSubscriptions(ctx, c.subName); err != nil {
 		return errors.Wrap(err, "failed to link subscription on startup")
 	}
diff --git a/services/load/service.go b/services/load/service.go
index 39518841..5028b913 100644
--- a/services/load/service.go
+++ b/services/load/service.go
@@ -12,7 +12,6 @@ import (
 	"sync"
 
 	"github.com/ghodss/yaml"
-
 	"github.com/influxdata/kapacitor/client/v1"
 	kexpvar "github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/server/vars"
diff --git a/services/smtp/service.go b/services/smtp/service.go
index c3bc7fa9..8cb01592 100644
--- a/services/smtp/service.go
+++ b/services/smtp/service.go
@@ -10,7 +10,6 @@ import (
 
 	"github.com/influxdata/kapacitor/alert"
 	"github.com/influxdata/kapacitor/keyvalue"
-
 	"gopkg.in/gomail.v2"
 )
 
diff --git a/services/storage/version_easyjson.go b/services/storage/version_easyjson.go
index ecfbc761..bcb42c45 100644
--- a/services/storage/version_easyjson.go
+++ b/services/storage/version_easyjson.go
@@ -4,7 +4,6 @@ package storage
 
 import (
 	json "encoding/json"
-
 	easyjson "github.com/mailru/easyjson"
 	jlexer "github.com/mailru/easyjson/jlexer"
 	jwriter "github.com/mailru/easyjson/jwriter"
diff --git a/services/teams/service.go b/services/teams/service.go
index eac08534..a330a2ea 100644
--- a/services/teams/service.go
+++ b/services/teams/service.go
@@ -6,12 +6,11 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"net/url"
 	"sync/atomic"
 
-	"math"
-
 	"github.com/influxdata/kapacitor/alert"
 	"github.com/influxdata/kapacitor/keyvalue"
 	"github.com/pkg/errors"
diff --git a/services/telegram/service.go b/services/telegram/service.go
index d0ef7cbd..a615a348 100644
--- a/services/telegram/service.go
+++ b/services/telegram/service.go
@@ -9,9 +9,8 @@ import (
 	"net/http"
 	"net/url"
 	"path"
-	"sync/atomic"
-
 	"strings"
+	"sync/atomic"
 
 	"github.com/influxdata/kapacitor/alert"
 	"github.com/influxdata/kapacitor/keyvalue"
diff --git a/task/backend/analytical_storage.go b/task/backend/analytical_storage.go
index a7f6f20b..7a982673 100644
--- a/task/backend/analytical_storage.go
+++ b/task/backend/analytical_storage.go
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/influxdata/flux"
@@ -29,6 +30,79 @@ const (
 	statusTag = "status"
 )
 
+func CreateDestination(log *zap.Logger, cli influxdb.Client, dest DataDestination) error {
+	const numTries = 5
+	for try := 0; ; try++ {
+		// poll for influxdb to come up - usually it should be already up by the time we get here.
+		if try > 0 {
+			time.Sleep(1 * time.Second)
+		}
+		queryForError := func(q string) error {
+			// This query is only successful if the bucket exists
+			result, err := cli.QueryFlux(influxdb.FluxQuery{
+				Org:   dest.Org,
+				OrgID: dest.OrgID,
+				Query: q,
+			})
+			if err != nil {
+				return err
+			}
+			for result.More() {
+				if err := result.Next().Tables().Do(func(table flux.Table) error {
+					return table.Do(func(col flux.ColReader) error {
+						return nil
+					})
+				}); err != nil {
+					return err
+				}
+			}
+			return result.Err()
+		}
+
+		// We do this instead of interpreting the response from `buckets()` because
+		// for 1.x `from(bucket: "mydb")` is valid and means to use the default retention
+		// policy, but `buckets()` will return the fully qualified retention policy.
+		err := queryForError(fmt.Sprintf(`from(bucket: %q) |> range(start: -1s) |> filter(fn: (r) => r._measurement == %q)`, dest.Bucket, dest.Measurement))
+		if err == nil {
+			log.Info("Successfully read from bucket for task analytic store")
+			return nil
+		}
+		log.Info("Error reading from bucket for task analytic storage, attempting to create it", zap.Error(err), zap.String("name", dest.Bucket))
+
+		if try > numTries {
+			break
+		}
+
+		// First check if we're reading any flux queries at all
+		err = queryForError("buckets()")
+		if err != nil {
+			log.Warn("InfluxDB instance for analytic store not processing flux queries", zap.Error(err))
+			continue
+		}
+
+		if strings.Contains(dest.Bucket, "/") {
+			log.Info("Bucket contains /, creating a 1.x database with non-default retention policy is not supported, skipping 1.x database creation", zap.String("name", dest.Bucket))
+		} else {
+			_, err := cli.Query(influxdb.Query{
+				Command: fmt.Sprintf("CREATE DATABASE %q", dest.Bucket),
+			})
+			if err != nil {
+				log.Warn("Error attempting to create 1.x database", zap.Error(err))
+			} else {
+				// we successfully tried to create a database
+				continue
+			}
+		}
+
+		log.Info("Could not create using 1.x CREATE DATABASE api, try 2.x /api/v2/bucket instead")
+		err = cli.CreateBucketV2(dest.Bucket, dest.Org, dest.OrgID)
+		if err != nil {
+			log.Warn("Error attempting to create 2.x bucket", zap.Error(err))
+		}
+	}
+	return fmt.Errorf("could not create database for analytic store, tried %d times", numTries)
+}
+
 // RunRecorder is a type which records runs into an influxdb
 // backed storage mechanism
 type RunRecorder interface {
@@ -41,22 +115,10 @@ type PointsWriter interface {
 
 type NoOpPointsWriter struct{}
 
-func (*NoOpPointsWriter) WritePoints(ctx context.Context, bucket string, points models.Points) error {
+func (*NoOpPointsWriter) WritePoints(_ context.Context, _ string, _ models.Points) error {
 	return nil
 }
 
-// NewAnalyticalStorage creates a new analytical store with access to the necessary systems for storing data and to act as a middleware (deprecated)
-func NewAnalyticalStorage(log *zap.Logger, ts taskmodel.TaskService, tcs TaskControlService, cli influxdb.Client, dest DataDestination) *AnalyticalStorage {
-	return &AnalyticalStorage{
-		log:                log,
-		TaskService:        ts,
-		TaskControlService: tcs,
-		destination:        dest,
-		rr:                 NewStoragePointsWriterRecorder(log, cli, dest),
-		cli:                cli,
-	}
-}
-
 type DataDestination struct {
 	Bucket      string
 	Org         string
@@ -74,6 +136,21 @@ type AnalyticalStorage struct {
 	log         *zap.Logger
 }
 
+// NewAnalyticalStorage creates a new analytical store with access to the necessary systems for storing data and to act as a middleware (deprecated)
+func NewAnalyticalStorage(log *zap.Logger, ts taskmodel.TaskService, tcs TaskControlService, cli influxdb.Client, dest DataDestination) (*AnalyticalStorage, error) {
+	if err := CreateDestination(log, cli, dest); err != nil {
+		return nil, err
+	}
+	return &AnalyticalStorage{
+		log:                log,
+		TaskService:        ts,
+		TaskControlService: tcs,
+		destination:        dest,
+		rr:                 NewStoragePointsWriterRecorder(log, cli, dest),
+		cli:                cli,
+	}, nil
+}
+
 func (as *AnalyticalStorage) FinishRun(ctx context.Context, taskID, runID platform.ID) (*taskmodel.Run, error) {
 	run, err := as.TaskControlService.FinishRun(ctx, taskID, runID)
 	if err != nil {
@@ -83,7 +160,7 @@ func (as *AnalyticalStorage) FinishRun(ctx context.Context, taskID, runID platfo
 }
 
 // FindLogs returns logs for a run.
-// First attempt to use the TaskService, then append additional analytical's logs to the list
+// First attempt to use the TaskService, then append additional analytical logs to the list
 func (as *AnalyticalStorage) FindLogs(ctx context.Context, filter taskmodel.LogFilter) ([]*taskmodel.Log, int, error) {
 	var logs []*taskmodel.Log
 	if filter.Run != nil {
@@ -113,7 +190,7 @@ func (as *AnalyticalStorage) FindLogs(ctx context.Context, filter taskmodel.LogF
 }
 
 // FindRuns returns a list of runs that match a filter and the total count of returned runs.
-// First attempt to use the TaskService, then append additional analytical's runs to the list
+// First attempt to use the TaskService, then append additional analytical runs to the list
 func (as *AnalyticalStorage) FindRuns(ctx context.Context, filter taskmodel.RunFilter) ([]*taskmodel.Run, int, error) {
 	if filter.Limit == 0 {
 		filter.Limit = taskmodel.TaskDefaultPageSize
diff --git a/task/config.go b/task/config.go
index 0eab67a4..1511640f 100644
--- a/task/config.go
+++ b/task/config.go
@@ -38,9 +38,12 @@ type Config struct {
 	Secrets map[string]string `toml:"secrets"`
 }
 
+const DefaultTaskRunBucket = "kapacitor_fluxtask_logs"
+
 func NewConfig() Config {
 	return Config{
 		TaskRunMeasurement: "runs",
+		TaskRunBucket:      DefaultTaskRunBucket,
 	}
 }
 
@@ -54,8 +57,5 @@ func (c Config) Validate() error {
 	if len(c.TaskRunOrgID) > 0 && len(c.TaskRunOrg) > 0 {
 		return fmt.Errorf("only one of task-run-org and task-run-orgid should be set")
 	}
-	if len(c.TaskRunBucket) == 0 {
-		return fmt.Errorf("task-run-bucket is required")
-	}
 	return nil
 }
diff --git a/tick/stack.go b/tick/stack.go
index 1ff8e010..f9a62eed 100644
--- a/tick/stack.go
+++ b/tick/stack.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	//"log"
 )
 
 var ErrEmptyStack = errors.New("stack is empty")
diff --git a/udf/agent/udf.pb.go b/udf/agent/udf.pb.go
index 3b9dc7f1..407ea8e1 100644
--- a/udf/agent/udf.pb.go
+++ b/udf/agent/udf.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.23.0
-// 	protoc        v3.11.4
+// 	protoc-gen-go v1.25.0
+// 	protoc        v3.11.1
 // source: udf.proto
 
 package agent


Test patch:
diff --git a/influxdb/client_test.go b/influxdb/client_test.go
index 114d5533..f7786844 100644
--- a/influxdb/client_test.go
+++ b/influxdb/client_test.go
@@ -13,6 +13,8 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestClient_Flux(t *testing.T) {
@@ -451,3 +453,118 @@ func TestBatchPoints_SettersGetters(t *testing.T) {
 		t.Errorf("Expected: %s, got %s", bp.WriteConsistency(), "wc2")
 	}
 }
+
+const cannedOrgResponse = `{
+        "links": {
+                "self": "/api/v2/orgs"
+        },
+        "orgs": [
+                {
+                        "links": {
+                                "buckets": "/api/v2/buckets?org=myorg",
+                                "dashboards": "/api/v2/dashboards?org=myorg",
+                                "labels": "/api/v2/orgs/e89732d2ac84d94a/labels",
+                                "logs": "/api/v2/orgs/e89732d2ac84d94a/logs",
+                                "members": "/api/v2/orgs/e89732d2ac84d94a/members",
+                                "owners": "/api/v2/orgs/e89732d2ac84d94a/owners",
+                                "secrets": "/api/v2/orgs/e89732d2ac84d94a/secrets",
+                                "self": "/api/v2/orgs/e89732d2ac84d94a",
+                                "tasks": "/api/v2/tasks?org=myorg"
+                        },
+                        "id": "e89732d2ac84d94a",
+                        "name": "org1",
+                        "description": "",
+                        "createdAt": "2021-09-13T18:57:23.816229Z",
+                        "updatedAt": "2021-09-13T18:57:23.816239Z"
+                },
+                {
+                        "links": {
+                                "buckets": "/api/v2/buckets?org=myorg",
+                                "dashboards": "/api/v2/dashboards?org=myorg",
+                                "labels": "/api/v2/orgs/0x0x0x0xac84d94a/labels",
+                                "logs": "/api/v2/orgs/0x0x0x0xac84d94a/logs",
+                                "members": "/api/v2/orgs/0x0x0x0xac84d94a/members",
+                                "owners": "/api/v2/orgs/0x0x0x0xac84d94a/owners",
+                                "secrets": "/api/v2/orgs/0x0x0x0xac84d94a/secrets",
+                                "self": "/api/v2/orgs/0x0x0x0xac84d94a",
+                                "tasks": "/api/v2/tasks?org=myorg"
+                        },
+                        "id": "0x0x0x0xac84d94a",
+                        "name": "org2",
+                        "description": "",
+                        "createdAt": "2021-09-13T18:57:23.816229Z",
+                        "updatedAt": "2021-09-13T18:57:23.816239Z"
+                }
+        ]
+}`
+
+func TestHTTPClient_CreateBucketV2(t *testing.T) {
+
+	tests := []struct {
+		name             string
+		org              string
+		orgId            string
+		bucket           string
+		expectBucketBody string
+		wantErr          string
+	}{
+		{
+			name:             "basic",
+			orgId:            "x0x0x0",
+			bucket:           "mybucket",
+			expectBucketBody: `{"orgID": "x0x0x0", "name": "mybucket", "retentionRules": [] }`,
+			wantErr:          "",
+		},
+		{
+			name:             "with-unknown-org",
+			org:              "some-other-org",
+			bucket:           "mybucket",
+			expectBucketBody: "",
+			wantErr:          "unknown organization name",
+		},
+		{
+			name:             "with-org",
+			org:              "org2",
+			bucket:           "mybucket",
+			expectBucketBody: `{"orgID": "0x0x0x0xac84d94a", "name": "mybucket", "retentionRules": [] }`,
+			wantErr:          "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotBucketBody := false
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				require.Equal(t, "Token mytoken", r.Header.Get("Authorization"))
+				body, err := ioutil.ReadAll(r.Body)
+				require.NoError(t, err)
+				switch r.URL.Path {
+				case "/api/v2/buckets":
+					assert.Equal(t, tt.expectBucketBody, string(body))
+					gotBucketBody = true
+					w.WriteHeader(201)
+				case "/api/v2/orgs":
+					w.WriteHeader(200)
+					w.Write([]byte(cannedOrgResponse))
+				default:
+					w.WriteHeader(404)
+				}
+			}))
+			c, err := NewHTTPClient(Config{
+				URLs: []string{ts.URL},
+				Credentials: Credentials{
+					Method: TokenAuthentication,
+					Token:  "mytoken",
+				},
+			})
+			require.NoError(t, err)
+			err = c.CreateBucketV2(tt.bucket, tt.org, tt.orgId)
+			if tt.wantErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.Contains(t, err.Error(), tt.wantErr)
+			}
+			assert.Equal(t, tt.expectBucketBody != "", gotBucketBody)
+		})
+	}
+}
diff --git a/integrations/streamer_test.go b/integrations/streamer_test.go
index 24766fe4..178beea7 100644
--- a/integrations/streamer_test.go
+++ b/integrations/streamer_test.go
@@ -9207,6 +9207,10 @@ stream
 			.value('{{ index .Fields "count" }}')
 			.services('serviceA', 'serviceB', '{{ .Name }}')
 			.correlated('{{ .Name }}')
+			.attribute('attributeA', '{{ .Name }}')
+			.attribute('attributeB', TRUE)
+			.attribute('attributeC', 9001.0)
+
 `
 	tmInit := func(tm *kapacitor.TaskMaster) {
 		c := alerta.NewConfig()
@@ -9231,6 +9235,7 @@ stream
 				Origin:      "Kapacitor",
 				Service:     []string{"cpu"},
 				Correlate:   []string{"cpu"},
+				Attributes:  nil,
 				Timeout:     3600,
 			},
 		},
@@ -9247,6 +9252,7 @@ stream
 				Service:     []string{"serviceA", "serviceB", "cpu"},
 				Correlate:   []string{"cpu"},
 				Value:       "10",
+				Attributes:  map[string]interface{}{"attributeA": "cpu", "attributeB": true, "attributeC": 9001.0},
 				Timeout:     86400,
 			},
 		},
diff --git a/pipeline/tick/alert_test.go b/pipeline/tick/alert_test.go
index 0d371091..a10790a0 100644
--- a/pipeline/tick/alert_test.go
+++ b/pipeline/tick/alert_test.go
@@ -667,6 +667,9 @@ func TestAlertAlerta(t *testing.T) {
 	handler.Services("legion", "vent", "garrus", "distraction team", "grunt", "crew", "samara", "barrier")
 	handler.Correlated("Harbinger")
 	handler.Timeout = 10 * time.Second
+	handler.Attribute("source", "Dragon Ball Z")
+	handler.Attribute("power", float64(9001))
+	handler.Attribute("o rly?", true)
 
 	want := `stream
     |from()
@@ -686,6 +689,9 @@ func TestAlertAlerta(t *testing.T) {
         .services('legion', 'vent', 'garrus', 'distraction team', 'grunt', 'crew', 'samara', 'barrier')
         .correlated('Harbinger')
         .timeout(10s)
+        .attribute('o rly?', TRUE)
+        .attribute('power', 9001.0)
+        .attribute('source', 'Dragon Ball Z')
 `
 	PipelineTickTestHelper(t, pipe, want)
 }
diff --git a/server/server_test.go b/server/server_test.go
index 0cd080c2..33e19c4f 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -9127,6 +9127,11 @@ func TestServer_ListServiceTests(t *testing.T) {
 						"testServiceX",
 						"testServiceY",
 					},
+					"attributes": map[string]interface{}{
+						"testAttributeA": "A",
+						"testAttributeB": true,
+						"testAttributeC": float64(9001.0),
+					},
 					"timeout": "24h0m0s",
 				},
 			},
@@ -10197,6 +10202,239 @@ stream
 	}
 }
 
+func TestServer_AlertJSON(t *testing.T) {
+	postServer := func(t *testing.T, expected string) *httptest.Server {
+		t.Helper()
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			data, err := io.ReadAll(r.Body)
+			if err != nil {
+				t.Error(err)
+			}
+			if (string(data)) != expected {
+				t.Error(
+					"Unexpected httppost request",
+					cmp.Diff(string(data), expected))
+			}
+		}))
+	}
+
+	testCases := []struct {
+		name     string
+		setup    func(*testing.T, *server.Config, string) (*httptest.Server, map[string]interface{}, error)
+		expected string
+		tick     string
+	}{
+		{
+			name: "normal_json",
+			setup: func(t *testing.T, c *server.Config, expected string) (*httptest.Server, map[string]interface{}, error) {
+				ts := postServer(t, expected)
+				ha := &client.TopicHandler{
+					Kind: "post",
+				}
+				ha.Options = map[string]interface{}{
+					"url":      ts.URL,
+					"endpoint": "test",
+				}
+				c.HTTPPost = httppost.Configs{{
+					Endpoint:      "test",
+					URLTemplate:   ts.URL,
+					Headers:       nil,
+					AlertTemplate: "{{ json . }}",
+				}}
+				return ts, ha.Options, nil
+			},
+			expected: `{"id":"id","message":"message","details":"details","time":"1970-01-01T00:00:00Z",` +
+				`"duration":0,"level":"CRITICAL","data":{"series":[{"name":"alert","columns":["time","value"],` +
+				`"values":[["1970-01-01T00:00:00Z",1]]}]},"previousLevel":"OK","recoverable":true}` +
+				"\n",
+			tick: `stream
+					|from()
+					.measurement('alert')
+					|alert()
+					.topic('test')
+					.id('id')
+					.message('message')
+					.details('details')
+					.crit(lambda: TRUE)`,
+		},
+		{
+			name: "compact-json",
+			setup: func(t *testing.T, c *server.Config, expected string) (*httptest.Server, map[string]interface{}, error) {
+				ts := postServer(t, expected)
+				ha := &client.TopicHandler{
+					Kind: "post",
+				}
+				ha.Options = map[string]interface{}{
+					"url":      ts.URL,
+					"endpoint": "test",
+				}
+				c.HTTPPost = httppost.Configs{{
+					Endpoint:      "test",
+					URLTemplate:   ts.URL,
+					Headers:       nil,
+					AlertTemplate: "{{ jsonCompact . }}", // note this gets ignored by the testing framework
+				}}
+				return ts, ha.Options, nil
+			},
+			expected: `{"id":"id","message":"message","details":"details","time":"1970-01-01T00:00:00Z","duration":0,"level":"CRITICAL","data":{"series":[{"name":"alert","columns":["time","value"],"values":[["1970-01-01T00:00:00Z",1]]}]},"previousLevel":"OK","recoverable":true}`,
+			tick: `stream
+			|from()
+			.measurement('alert')
+			|alert()
+			.topic('test')
+			.id('id')
+			.message('message')
+			.details('details')
+			.crit(lambda: TRUE)`,
+		},
+		{
+			name: "details_normal_json",
+			setup: func(t *testing.T, c *server.Config, expected string) (*httptest.Server, map[string]interface{}, error) {
+				ts := postServer(t, expected)
+				ha := &client.TopicHandler{
+					Kind: "post",
+				}
+				ha.Options = map[string]interface{}{
+					"url":      ts.URL,
+					"endpoint": "test",
+				}
+				c.HTTPPost = httppost.Configs{{
+					Endpoint:      "test",
+					URLTemplate:   ts.URL,
+					Headers:       nil,
+					AlertTemplate: "{{ json . }}",
+				}}
+				return ts, ha.Options, nil
+			},
+			expected: `{"id":"id","message":"message","details":` +
+				`"{\u0026#34;Name\u0026#34;:\u0026#34;alert\u0026#34;,\u0026#34;Ta` +
+				`skName\u0026#34;:\u0026#34;testAlertHandlers\u0026#34;,\u0026#34` +
+				`;Group\u0026#34;:\u0026#34;nil\u0026#34;,\u0026#34;Tags\u0026#34` +
+				`;:{},\u0026#34;ServerInfo\u0026#34;:{\u0026#34;Hostname\u0026#34` +
+				`;:\u0026#34;localhost\u0026#34;,\u0026#34;ClusterID\u0026#34;:\u` +
+				`0026#34;4c77777-a43d-439a-ba56-b08cbb0993fe\u0026#34;,\u0026#34` +
+				`;ServerID\u0026#34;:\u0026#34;af80a060-3226-41e2-b1cd-548a1467b491` +
+				`91\u0026#34;},\u0026#34;ID\u0026#34;:\u0026#34;id\u0026#34;,\u00` +
+				`26#34;Fields\u0026#34;:{\u0026#34;value\u0026#34;:1},\u0026#34;L` +
+				`evel\u0026#34;:\u0026#34;CRITICAL\u0026#34;,\u0026#34;Time\u0026` +
+				`#34;:\u0026#34;1970-01-01T00:00:00Z\u0026#34;,\u0026#34;Duration` +
+				`\u0026#34;:0,\u0026#34;Message\u0026#34;:\u0026#34;message\u0026` +
+				`#34;}\n",` +
+				`"time":"1970-01-01T00:00:00Z","duration":0,"level":"CRITICAL",` +
+				`"data":{"series":[{"name":"alert","columns":["time","value"],` +
+				`"values":[["1970-01-01T00:00:00Z",1]]}]},"previousLevel":"OK","recoverable":true}`,
+			tick: `stream
+			|from()
+			.measurement('alert')
+			|alert()
+			.topic('test')
+			.id('id')
+			.message('message')
+			.details('{{ json .data }}')
+			.crit(lambda: TRUE)`,
+		},
+		{
+			name: "details_compact_json",
+			setup: func(t *testing.T, c *server.Config, expected string) (*httptest.Server, map[string]interface{}, error) {
+				ts := postServer(t, expected)
+				ha := &client.TopicHandler{
+					Kind: "post",
+				}
+				ha.Options = map[string]interface{}{
+					"url":      ts.URL,
+					"endpoint": "test",
+				}
+				c.HTTPPost = httppost.Configs{{
+					Endpoint:      "test",
+					URLTemplate:   ts.URL,
+					Headers:       nil,
+					AlertTemplate: "{{ json . }}",
+				}}
+				return ts, ha.Options, nil
+			},
+			expected: `{"id":"id","message":"message","details":` +
+				`"{\u0026#34;Name\u0026#34;:\u0026#34;alert\u0026#34;,\u0026#34;Ta` +
+				`skName\u0026#34;:\u0026#34;testAlertHandlers\u0026#34;,\u0026#34` +
+				`;Group\u0026#34;:\u0026#34;nil\u0026#34;,\u0026#34;Tags\u0026#34` +
+				`;:{},\u0026#34;ServerInfo\u0026#34;:{\u0026#34;Hostname\u0026#34` +
+				`;:\u0026#34;localhost\u0026#34;,\u0026#34;ClusterID\u0026#34;:\u` +
+				`0026#34;f4c77777-a43d-439a-ba56-b08cbb0993fe\u0026#34;,\u0026#34` +
+				`;ServerID\u0026#34;:\u0026#34;af80a060-3226-41e2-b1cd-548a1467b4` +
+				`91\u0026#34;},\u0026#34;ID\u0026#34;:\u0026#34;id\u0026#34;,\u00` +
+				`26#34;Fields\u0026#34;:{\u0026#34;value\u0026#34;:1},\u0026#34;L` +
+				`evel\u0026#34;:\u0026#34;CRITICAL\u0026#34;,\u0026#34;Time\u0026` +
+				`#34;:\u0026#34;1970-01-01T00:00:00Z\u0026#34;,\u0026#34;Duration` +
+				`\u0026#34;:0,\u0026#34;Message\u0026#34;:\u0026#34;message\u0026` +
+				`#34;},` +
+				`"time":"1970-01-01T00:00:00Z","duration":0,"level":"CRITICAL",` +
+				`"data":{"series":[{"name":"alert","columns":["time","value"],` +
+				`"values":[["1970-01-01T00:00:00Z",1]]}]},"previousLevel":"OK","recoverable":true}` +
+				"\n",
+			tick: `stream
+			|from()
+			.measurement('alert')
+			|alert()
+			.topic('test')
+			.id('id')
+			.message('message')
+			.details('{{ jsonCompact .data }}')
+			.crit(lambda: TRUE)`,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+
+			// Create default config
+			c := NewConfig()
+			var options map[string]interface{}
+			ctxt := context.Background()
+			var ts *httptest.Server
+			var err error
+			ts, options, err = tc.setup(t, c, tc.expected)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer ts.Close()
+			s := OpenServer(c)
+			cli := Client(s)
+			defer func() {
+				s.Close()
+			}()
+			ctxt = context.WithValue(ctxt, "kapacitorURL", s.URL())
+
+			_, err = cli.CreateTopicHandler(cli.TopicHandlersLink("test"), client.TopicHandlerOptions{
+				ID:      "testJSONHandlers",
+				Kind:    "post",
+				Options: options,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if _, err := cli.CreateTask(client.CreateTaskOptions{
+				ID:   "testAlertHandlers",
+				Type: client.StreamTask,
+				DBRPs: []client.DBRP{{
+					Database:        "mydb",
+					RetentionPolicy: "myrp",
+				}},
+				TICKscript: tc.tick,
+				Status:     client.Enabled,
+			}); err != nil {
+				t.Fatalf("%s: %v", tc.name, err)
+			}
+
+			point := "alert value=1 0000000000"
+			v := url.Values{}
+			v.Add("precision", "s")
+			s.MustWrite("mydb", "myrp", point, v)
+
+		})
+	}
+
+}
+
 func TestServer_AlertHandlers(t *testing.T) {
 
 	resultJSON := `{"series":[{"name":"alert","columns":["time","value"],"values":[["1970-01-01T00:00:00Z",1]]}]}`
diff --git a/services/alerta/alertatest/alertatest.go b/services/alerta/alertatest/alertatest.go
index fbb22cd8..a1acdad6 100644
--- a/services/alerta/alertatest/alertatest.go
+++ b/services/alerta/alertatest/alertatest.go
@@ -53,14 +53,15 @@ type Request struct {
 }
 
 type PostData struct {
-	Resource    string   `json:"resource"`
-	Event       string   `json:"event"`
-	Group       string   `json:"group"`
-	Environment string   `json:"environment"`
-	Text        string   `json:"text"`
-	Origin      string   `json:"origin"`
-	Service     []string `json:"service"`
-	Correlate   []string `json:"correlate"`
-	Value       string   `json:"value"`
-	Timeout     int64    `json:"timeout"`
+	Resource    string                 `json:"resource"`
+	Event       string                 `json:"event"`
+	Group       string                 `json:"group"`
+	Environment string                 `json:"environment"`
+	Text        string                 `json:"text"`
+	Origin      string                 `json:"origin"`
+	Service     []string               `json:"service"`
+	Correlate   []string               `json:"correlate"`
+	Attributes  map[string]interface{} `json:"attributes"`
+	Value       string                 `json:"value"`
+	Timeout     int64                  `json:"timeout"`
 }
diff --git a/services/bigpanda/service_test.go b/services/bigpanda/service_test.go
index 526730c0..b7a88872 100644
--- a/services/bigpanda/service_test.go
+++ b/services/bigpanda/service_test.go
@@ -2,10 +2,12 @@ package bigpanda
 
 import (
 	"bytes"
-	"github.com/influxdata/kapacitor/alert"
-	"github.com/influxdata/kapacitor/models"
+	"fmt"
 	"testing"
 	"time"
+
+	"github.com/influxdata/kapacitor/alert"
+	"github.com/influxdata/kapacitor/models"
 )
 
 func TestService_SerializeEventData(t *testing.T) {
@@ -21,58 +23,60 @@ func TestService_SerializeEventData(t *testing.T) {
 	}{
 		{
 			fields:  map[string]interface{}{"primitive_type": 10},
-			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"10\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"10\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 		{
 			fields:  map[string]interface{}{"primitive_type": "string"},
-			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"string\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"string\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 		{
 			fields:  map[string]interface{}{"primitive_type": true},
-			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"true\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"true\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 		{
 			fields:  map[string]interface{}{"primitive_type": 123.45},
-			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"123.45\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"primitive_type\":\"123.45\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 		{
 			fields:  map[string]interface{}{"escape": "\n"},
-			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"escape\":\"\\n\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"escape\":\"\\n\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 		{
 			fields:  map[string]interface{}{"array": []interface{}{10, true, "string value"}},
-			expBody: "{\"app_key\":\"key\",\"array\":\"[10,true,\\\"string value\\\"]\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}\n",
+			expBody: "{\"app_key\":\"key\",\"array\":\"[10,true,\\\"string value\\\"]\",\"check\":\"id\",\"description\":\"message\",\"details\":\"details\",\"status\":\"ok\",\"task\":\":test\",\"timestamp\":31536038}",
 		},
 	}
 
-	for _, tc := range testCases {
-		r, err := s.preparePost(
-			"id",
-			"message",
-			"details",
-			alert.OK,
-			time.Date(1971, 1, 1, 0, 0, 38, 0, time.UTC),
-			alert.EventData{
-				Name:   "test",
-				Tags:   make(map[string]string),
-				Fields: tc.fields,
-				Result: models.Result{},
-			},
-			&HandlerConfig{})
+	for i, tc := range testCases {
+		t.Run(fmt.Sprint(i), func(t *testing.T) {
+			r, err := s.preparePost(
+				"id",
+				"message",
+				"details",
+				alert.OK,
+				time.Date(1971, 1, 1, 0, 0, 38, 0, time.UTC),
+				alert.EventData{
+					Name:   "test",
+					Tags:   make(map[string]string),
+					Fields: tc.fields,
+					Result: models.Result{},
+				},
+				&HandlerConfig{})
 
-		if err != nil {
-			t.Fatal(err)
-		}
+			if err != nil {
+				t.Fatal(err)
+			}
 
-		buf := new(bytes.Buffer)
-		_, err = buf.ReadFrom(r.Body)
-		if err != nil {
-			t.Fatal(err)
-		}
-		newStr := buf.String()
+			buf := new(bytes.Buffer)
+			_, err = buf.ReadFrom(r.Body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			newStr := buf.String()
 
-		if tc.expBody != newStr {
-			t.Fatalf("unexpected content: got '%s' exp '%s'", newStr, tc.expBody)
-		}
+			if tc.expBody != newStr {
+				t.Fatalf("unexpected content: got '%s' exp '%s'", newStr, tc.expBody)
+			}
+		})
 	}
 }
diff --git a/services/influxdb/service_test.go b/services/influxdb/service_test.go
index 87b9bdc0..5a76850c 100644
--- a/services/influxdb/service_test.go
+++ b/services/influxdb/service_test.go
@@ -1303,6 +1303,10 @@ func (c influxDBClient) Update(config influxcli.Config) error {
 	return nil
 }
 
+func (c influxDBClient) CreateBucketV2(bucket string, org string, orgID string) error {
+	return nil
+}
+
 type logSerivce struct {
 }
 
diff --git a/test.sh b/test.sh
index c991bd90..bf167d46 100755
--- a/test.sh
+++ b/test.sh
@@ -35,7 +35,7 @@ NO_UNCOMMITTED=${NO_UNCOMMITTED-false}
 # Home dir of the docker user
 HOME_DIR=/root
 # Go version
-GO_VERSION=1.17
+GO_VERSION=1.17.1
 # GOPATH
 GOPATH=/go