Repository: influxdata/kapacitor
Pull request #2523 (2021-04-02 20:36:20)

chore: merge the 1.6 branch back into master
chore: merge the 1.6 branch back into master


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ae2f9d04..d932885a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,12 @@
 # Changelog
 
+## Unreleased
+
+### Features
+- [#2484](https://github.com/influxdata/kapacitor/pull/2484): Add Zenoss alert event handler.
+- [#2512](https://github.com/influxdata/kapacitor/pull/2512): Pull in auth code from Kapacitor Enterprise.
+- [#2493](https://github.com/influxdata/kapacitor/pull/2493): Route kafka alerts to partitions by ID, and allow for configuring the hashing strategy.
+
 ## v1.5.9 [2021-04-01]
 
 ### Bugfixes
@@ -1447,5 +1454,3 @@ Some bug fixes including one that cause Kapacitor to deadlock.
 ### Release Notes
 
 Major public release.
-
-
diff --git a/alert.go b/alert.go
index 765737b3..7c7f0aba 100644
--- a/alert.go
+++ b/alert.go
@@ -36,6 +36,7 @@ import (
 	"github.com/influxdata/kapacitor/services/teams"
 	"github.com/influxdata/kapacitor/services/telegram"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/influxdata/kapacitor/tick/ast"
 	"github.com/influxdata/kapacitor/tick/stateful"
 	"github.com/pkg/errors"
@@ -338,9 +339,11 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 
 	for _, k := range n.KafkaHandlers {
 		c := kafka.HandlerConfig{
-			Cluster:  k.Cluster,
-			Topic:    k.KafkaTopic,
-			Template: k.Template,
+			Cluster:              k.Cluster,
+			Topic:                k.KafkaTopic,
+			Template:             k.Template,
+			DisablePartitionById: k.IsDisablePartitionById,
+			PartitionAlgorithm:   k.PartitionHashAlgorithm,
 		}
 		h, err := et.tm.KafkaService.Handler(c, ctx...)
 		if err != nil {
@@ -576,6 +579,36 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 		n.IsStateChangesOnly = true
 	}
 
+	for _, s := range n.ZenossHandlers {
+		c := zenoss.HandlerConfig{
+			URL:           s.URL,
+			Action:        s.Action,
+			Method:        s.Method,
+			Type:          s.Type,
+			TID:           s.TID,
+			Summary:       s.Summary,
+			Device:        s.Device,
+			Component:     s.Component,
+			EventClassKey: s.EventClassKey,
+			EventClass:    s.EventClass,
+			Message:       s.Message,
+			Collector:     s.Collector,
+			CustomFields:  s.CustomFieldsMap,
+		}
+		h := et.tm.ZenossService.Handler(c, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+	if len(n.ZenossHandlers) == 0 && (et.tm.ZenossService != nil && et.tm.ZenossService.Global()) {
+		h := et.tm.ZenossService.Handler(zenoss.HandlerConfig{}, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+	// If zenoss has been configured with state changes only set it.
+	if et.tm.ZenossService != nil &&
+		et.tm.ZenossService.Global() &&
+		et.tm.ZenossService.StateChangesOnly() {
+		n.IsStateChangesOnly = true
+	}
+
 	// Parse level expressions
 	an.levels = make([]stateful.Expression, alert.Critical+1)
 	an.scopePools = make([]stateful.ScopePool, alert.Critical+1)
diff --git a/client/API.md b/client/API.md
index 2201efa1..f573be5e 100644
--- a/client/API.md
+++ b/client/API.md
@@ -1407,6 +1407,10 @@ GET /kapacitor/v1/replays
 }
 ```
 
+## Users
+
+This section is under construction.
+
 ## Alerts
 
 Kapacitor can generate and handle alerts.
diff --git a/client/v1/client.go b/client/v1/client.go
index da5f7673..3f85b066 100644
--- a/client/v1/client.go
+++ b/client/v1/client.go
@@ -45,6 +45,7 @@ const (
 	replaysPath       = basePath + "/replays"
 	replayBatchPath   = basePath + "/replays/batch"
 	replayQueryPath   = basePath + "/replays/query"
+	usersPath         = basePath + "/users"
 	configPath        = basePath + "/config"
 	serviceTestsPath  = basePath + "/service-tests"
 	alertsPath        = basePath + "/alerts"
@@ -56,6 +57,276 @@ const (
 	backupPath        = storagePath + "/backup"
 )
 
+type UserType int
+
+const (
+	InvalidUser UserType = iota
+	NormalUser
+	AdminUser
+)
+
+func (ut UserType) MarshalText() ([]byte, error) {
+	switch ut {
+	case InvalidUser:
+		return []byte{}, nil
+	case NormalUser:
+		return []byte("normal"), nil
+	case AdminUser:
+		return []byte("admin"), nil
+	default:
+		return nil, fmt.Errorf("unknown UserType %d", ut)
+	}
+}
+
+func (ut *UserType) UnmarshalText(text []byte) error {
+	switch s := string(text); s {
+	case "":
+		*ut = InvalidUser
+	case "normal":
+		*ut = NormalUser
+	case "admin":
+		*ut = AdminUser
+	default:
+		return fmt.Errorf("unknown UserType %s", s)
+	}
+	return nil
+}
+
+func (ut UserType) String() string {
+	s, err := ut.MarshalText()
+	if err != nil {
+		return err.Error()
+	}
+	return string(s)
+}
+
+type Permission int
+
+const (
+	NoPermissions Permission = iota
+
+	APIPermission
+	ConfigAPIPermission
+	WritePointsPermission
+
+	AllPermissions
+)
+
+func (p Permission) MarshalText() ([]byte, error) {
+	switch p {
+	case NoPermissions:
+		return []byte("none"), nil
+	case APIPermission:
+		return []byte("api"), nil
+	case ConfigAPIPermission:
+		return []byte("config_api"), nil
+	case WritePointsPermission:
+		return []byte("write_points"), nil
+	case AllPermissions:
+		return []byte("all"), nil
+	default:
+		return nil, fmt.Errorf("unknown Permission %d", p)
+	}
+}
+
+func (p *Permission) UnmarshalText(text []byte) error {
+	switch s := string(text); s {
+	case "none":
+		*p = NoPermissions
+	case "api":
+		*p = APIPermission
+	case "config_api":
+		*p = ConfigAPIPermission
+	case "write_points":
+		*p = WritePointsPermission
+	case "all":
+		*p = AllPermissions
+	default:
+		return fmt.Errorf("unknown Permission  %s", s)
+	}
+	return nil
+}
+
+func (p Permission) String() string {
+	s, err := p.MarshalText()
+	if err != nil {
+		return err.Error()
+	}
+	return string(s)
+}
+
+type CreateUserOptions struct {
+	Name        string       `json:"name"`
+	Password    string       `json:"password"`
+	Type        UserType     `json:"type"`
+	Permissions []Permission `json:"permissions"`
+}
+
+type UpdateUserOptions struct {
+	Password    string       `json:"password,omitempty"`
+	Type        UserType     `json:"type,omitempty"`
+	Permissions []Permission `json:"permissions"` // NOTE: do not set omitempty, so we can distingush unset vs empty.
+}
+
+type User struct {
+	Link        Link         `json:"link"`
+	Name        string       `json:"name"`
+	Type        UserType     `json:"type"`
+	Permissions []Permission `json:"permissions"`
+}
+
+func (c *Client) UserLink(username string) Link {
+	return Link{Relation: Self, Href: path.Join(usersPath, username)}
+}
+
+// Create a new user.
+// Errors if the user already exists.
+func (c *Client) CreateUser(opt CreateUserOptions) (User, error) {
+	user := User{}
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	err := enc.Encode(opt)
+	if err != nil {
+		return user, err
+	}
+
+	u := c.BaseURL()
+	u.Path = usersPath
+
+	req, err := http.NewRequest("POST", u.String(), &buf)
+	if err != nil {
+		return user, err
+	}
+
+	_, err = c.Do(req, &user, http.StatusOK)
+	return user, err
+}
+
+// Update an existing user.
+// Only fields that are not their default value will be updated.
+// An empty list of permissions is different from a nil list of permissions.
+// An empty list updates the user to have no permissions, while an nil list of permissions leaves the user's permissions unmodified.
+func (c *Client) UpdateUser(link Link, opt UpdateUserOptions) (User, error) {
+	user := User{}
+	if link.Href == "" {
+		return user, fmt.Errorf("invalid link %v", link)
+	}
+
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	err := enc.Encode(opt)
+	if err != nil {
+		return user, err
+	}
+
+	u := c.BaseURL()
+	u.Path = link.Href
+
+	req, err := http.NewRequest("PATCH", u.String(), &buf)
+	if err != nil {
+		return user, err
+	}
+
+	_, err = c.Do(req, &user, http.StatusOK)
+	if err != nil {
+		return user, err
+	}
+	return user, nil
+}
+
+// Get information about a user.
+func (c *Client) User(link Link) (User, error) {
+	user := User{}
+	if link.Href == "" {
+		return user, fmt.Errorf("invalid link %v", link)
+	}
+
+	u := c.BaseURL()
+	u.Path = link.Href
+
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return user, err
+	}
+
+	_, err = c.Do(req, &user, http.StatusOK)
+	if err != nil {
+		return user, err
+	}
+	return user, nil
+}
+
+// Delete a user.
+func (c *Client) DeleteUser(link Link) error {
+	if link.Href == "" {
+		return fmt.Errorf("invalid link %v", link)
+	}
+
+	u := c.BaseURL()
+	u.Path = link.Href
+
+	req, err := http.NewRequest("DELETE", u.String(), nil)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.Do(req, nil, http.StatusNoContent)
+	return err
+}
+
+type ListUsersOptions struct {
+	Pattern string
+	Fields  []string
+	Offset  int
+	Limit   int
+}
+
+func (o *ListUsersOptions) Default() {
+	if o.Limit == 0 {
+		o.Limit = 100
+	}
+}
+
+func (o *ListUsersOptions) Values() *url.Values {
+	v := &url.Values{}
+	v.Set("pattern", o.Pattern)
+	for _, field := range o.Fields {
+		v.Add("fields", field)
+	}
+	v.Set("offset", strconv.FormatInt(int64(o.Offset), 10))
+	v.Set("limit", strconv.FormatInt(int64(o.Limit), 10))
+	return v
+}
+
+// Get users.
+func (c *Client) ListUsers(opt *ListUsersOptions) ([]User, error) {
+	if opt == nil {
+		opt = new(ListUsersOptions)
+	}
+	opt.Default()
+	u := c.BaseURL()
+	u.Path = usersPath
+	u.RawQuery = opt.Values().Encode()
+
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	// Response type
+	type response struct {
+		Users []User `json:"users"`
+	}
+
+	r := &response{}
+
+	_, err = c.Do(req, r, http.StatusOK)
+	if err != nil {
+		return nil, err
+	}
+	return r.Users, nil
+}
+
 // HTTP configuration for connecting to Kapacitor
 type Config struct {
 	// The URL of the Kapacitor server.
diff --git a/etc/kapacitor/kapacitor.conf b/etc/kapacitor/kapacitor.conf
index 979011c5..fb22d970 100644
--- a/etc/kapacitor/kapacitor.conf
+++ b/etc/kapacitor/kapacitor.conf
@@ -353,6 +353,26 @@ default-retention-policy = ""
   # The URL for the Pushover API.
   url = "https://api.pushover.net/1/messages.json"
 
+[zenoss]
+  # Configure Zenoss.
+  enabled = false
+  # The Zenoss events endpoint URL.
+  url = "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router"
+  # Username for HTTP BASIC authentication
+  # username = ""
+  # Password for HTTP BASIC authentication
+  # password = ""
+  # Action (router name).
+  action = "EventsRouter"
+  # Router method.
+  method = "add_event"
+  # Event type.
+  type = "rpc"
+  # Temporary request transaction ID.
+  tid = 1
+  # Alert level to event severity mapping.
+  severity-map = { OK = "Clear", Info = "Info", Warning = "Warning", Critical = "Critical" }
+
 ##########################################
 # Configure Alert POST request Endpoints
 
diff --git a/go.mod b/go.mod
index 14815d9b..deee16c4 100644
--- a/go.mod
+++ b/go.mod
@@ -81,7 +81,7 @@ require (
 	github.com/russross/blackfriday v0.0.0-20151110051855-0b647d0506a6 // indirect
 	github.com/samuel/go-zookeeper v0.0.0-20161028232340-1d7be4effb13 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
-	github.com/segmentio/kafka-go v0.3.5
+	github.com/segmentio/kafka-go v0.3.10
 	github.com/serenize/snaker v0.0.0-20161123064335-543781d2b79b
 	github.com/shurcooL/go v0.0.0-20170331015642-20b4b0a35211 // indirect
 	github.com/shurcooL/markdownfmt v0.0.0-20170214213350-10aae0a270ab
@@ -93,6 +93,7 @@ require (
 	github.com/syndtr/goleveldb v0.0.0-20170409015612-8c81ea47d4c4 // indirect
 	github.com/ugorji/go v0.0.0-20170312112114-708a42d24682 // indirect
 	go.etcd.io/bbolt v1.3.5
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
 	golang.org/x/net v0.0.0-20201021035429-f5854403a974 // indirect
 	golang.org/x/oauth2 v0.0.0-20170412232759-a6bd8cefa181 // indirect
 	golang.org/x/sys v0.0.0-20210305215415-5cdee2b1b5a0 // indirect
diff --git a/go.sum b/go.sum
index 9438712a..461c369f 100644
--- a/go.sum
+++ b/go.sum
@@ -8,8 +8,6 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
-github.com/DataDog/zstd v1.4.0 h1:vhoV+DUHnRZdKW1i5UMjAk2G4JY8wN4ayRfYDNdEhwo=
-github.com/DataDog/zstd v1.4.0/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
 github.com/Masterminds/semver v1.4.2 h1:WBLTQ37jOCzSLtXNdoo8bNM8876KhNqOKvrlGITgsTc=
 github.com/Masterminds/semver v1.4.2/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
 github.com/Masterminds/sprig v2.16.0+incompatible h1:QZbMUPxRQ50EKAq3LFMnxddMu88/EUUG3qmxwtDmPsY=
@@ -43,7 +41,6 @@ github.com/blang/semver v3.5.0+incompatible h1:CGxCgetQ64DKk7rdZ++Vfnb1+ogGNnB17
 github.com/blang/semver v3.5.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40 h1:y4B3+GPxKlrigF1ha5FFErxK+sr6sWxQovRMzwMhejo=
 github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+WjiTceGBHIoTvn60HIbs7Hm7bcHjyrSqYB9c=
-github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/cenkalti/backoff v1.0.0 h1:2XeuDgvPv/6QDyzIuxb6n36ADVocyqTLlOSpYBGYtvM=
 github.com/cenkalti/backoff v1.0.0/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
@@ -55,7 +52,6 @@ github.com/coreos/go-oidc v0.0.0-20170307191026-be73733bb8cc h1:9yuvA19Q5WFkLwJc
 github.com/coreos/go-oidc v0.0.0-20170307191026-be73733bb8cc/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc=
 github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf h1:CAKfRE2YtTUIjjh1bkBtyYFaUT/WmOqsJjgtihT0vMI=
 github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -119,7 +115,6 @@ github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:x
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
 github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
-github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
@@ -129,7 +124,6 @@ github.com/goller/prometheus v1.6.1-0.20170502220046-58298e738211 h1:uckek0Fbc11
 github.com/goller/prometheus v1.6.1-0.20170502220046-58298e738211/go.mod h1:h9lxWAIUxz0+oYuWHSf3a8m22+Sc9ZXJ3htO8x9b9OM=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -208,6 +202,8 @@ github.com/k-sone/snmpgo v3.2.0+incompatible h1:2NogYilKYSia0f+seO9P7aRa6MKG6Rcn
 github.com/k-sone/snmpgo v3.2.0+incompatible/go.mod h1:9MC6LeG1sGPgrwnmu/V/ncg9P2M5zS5IvE+c4KZj25g=
 github.com/kimor79/gollectd v1.0.0 h1:6APoEersLJ2W0iwWj8C7COlienQ8XTWh8np4w0ggI8k=
 github.com/kimor79/gollectd v1.0.0/go.mod h1:lDxzEAixH34FPZ0nBIpjCu2vR3ZdIKZbbnf2rc+b2ao=
+github.com/klauspost/compress v1.9.8 h1:VMAMUUOh+gaxKTMk+zqbjsSjsIcUcL/LF4o63i82QyA=
+github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
@@ -256,7 +252,6 @@ github.com/pborman/uuid v0.0.0-20160209185913-a97ce2ca70fa h1:l8VQbMdmwFH37kOOaW
 github.com/pborman/uuid v0.0.0-20160209185913-a97ce2ca70fa/go.mod h1:VyrYX9gd7irzKovcSS6BIIEwPRkP2Wm2m9ufcdFSJ34=
 github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
-github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -288,8 +283,8 @@ github.com/samuel/go-zookeeper v0.0.0-20161028232340-1d7be4effb13 h1:4AQBn5RJY4W
 github.com/samuel/go-zookeeper v0.0.0-20161028232340-1d7be4effb13/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
-github.com/segmentio/kafka-go v0.3.5 h1:2JVT1inno7LxEASWj+HflHh5sWGfM0gkRiLAxkXhGG4=
-github.com/segmentio/kafka-go v0.3.5/go.mod h1:OT5KXBPbaJJTcvokhWR2KFmm0niEx3mnccTwjmLvSi4=
+github.com/segmentio/kafka-go v0.3.10 h1:h/1aSu7gWp6DXLmp0csxm8wrYD6rRYyaqclu2aQ/PWo=
+github.com/segmentio/kafka-go v0.3.10/go.mod h1:8rEphJEczp+yDE/R5vwmaqZgF1wllrl4ioQcNKB8wVA=
 github.com/serenize/snaker v0.0.0-20161123064335-543781d2b79b h1:B6dcIy62mIVH3xZ+Alc5J4fLDWthEa1RPzK7L7/glTw=
 github.com/serenize/snaker v0.0.0-20161123064335-543781d2b79b/go.mod h1:Yow6lPLSAXx2ifx470yD/nUe22Dv5vBvxK/UK9UUTVs=
 github.com/shurcooL/go v0.0.0-20170331015642-20b4b0a35211 h1:+OTcQECFnrdh6oabESXRUNFuM3Ol347YZvD+sL/Nfv0=
@@ -329,7 +324,6 @@ go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 golang.org/x/crypto v0.0.0-20180505025534-4ec37c66abab/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190506204251-e1dfcc566284 h1:rlLehGeYg6jfoyz/eDqDU1iRXLKfR42nnNh57ytKEWo=
 golang.org/x/crypto v0.0.0-20190506204251-e1dfcc566284/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -337,11 +331,9 @@ golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
-golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0 h1:wBouT66WTYFXdxfVdz9sVWARVd/2vfGcmI45D2gj45M=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974 h1:IX6qOQeG5uLjB/hjjwjedwfjND0hgjPMMyO1RoIXQNI=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
@@ -356,7 +348,6 @@ golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190523142557-0e01d883c5c5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -367,11 +358,9 @@ golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210305215415-5cdee2b1b5a0 h1:MOJR6AyRlIYMexU2acorBot1aPks0cBDOyUA4hFlBhE=
 golang.org/x/sys v0.0.0-20210305215415-5cdee2b1b5a0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
@@ -379,7 +368,6 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190424220101-1e8e1cfdf96b/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/pipeline/alert.go b/pipeline/alert.go
index cb7867bf..fc51cd0c 100644
--- a/pipeline/alert.go
+++ b/pipeline/alert.go
@@ -404,6 +404,10 @@ type AlertNodeData struct {
 	// Send alert to ServiceNow.
 	// tick:ignore
 	ServiceNowHandlers []*ServiceNowHandler `tick:"ServiceNow" json:"serviceNow"`
+
+	// Send alert to Zenoss.
+	// tick:ignore
+	ZenossHandlers []*ZenossHandler `tick:"Zenoss" json:"zenoss"`
 }
 
 func newAlertNode(wants EdgeType) *AlertNode {
@@ -2155,9 +2159,33 @@ type KafkaHandler struct {
 	// Kafka Topic
 	KafkaTopic string `json:"kafka-topic"`
 
+	// If false (default), messages will be routed to partitions based on their IDs
+	// If true, messages will be routed to the partition with the least data regardless of IDs
+	// tick:ignore
+	IsDisablePartitionById bool `tick:"DisablePartitionById" json:"disable-partition-by-id"`
+
+	// Algorithm used to hash message IDs when determining a target partition
+	//
+	// Valid values are:
+	//
+	//    * "crc32"   - Compatible with librdkafka and confluent-kafka-go
+	//    * "murmur2" - Compatible with the default Java partitioner
+	//    * "fnv-1a"  - Compatible with Shopify's sarama producer
+	//
+	// Default: crc32
+	PartitionHashAlgorithm string `json:"partition-hash-algorithm,omitempty"`
+
 	// Template used to construct the message body
 	// If empty the alert data in JSON is sent as the message body.
-	Template string `json:"template"`
+	// tick:ignore
+	Template string `json:"template,omitempty"`
+}
+
+// Disables use of message IDs when determining target Kafka partitions.
+// tick:property
+func (k *KafkaHandler) DisablePartitionById() *KafkaHandler {
+	k.IsDisablePartitionById = true
+	return k
 }
 
 // Send the alert to a Microsoft Teams channel.
@@ -2313,3 +2341,120 @@ func (s *ServiceNowHandler) AdditionalInfo(key string, value interface{}) *Servi
 	s.AdditionalInfoMap[key] = value
 	return s
 }
+
+// Send the alert to Zenoss.
+//
+// Example:
+//    [zenoss]
+//      enabled = true
+//      url = "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router"
+//
+// In order to not post a message every alert interval
+// use AlertNode.StateChangesOnly so that only events
+// where the alert changed state are posted.
+//
+// Example:
+//    stream
+//         |alert()
+//             .zenoss()
+//                 .summary('Alert {{ .ID }}')
+//                 .message('{{ .Message }}')
+//                 .eventClass('/App')
+//
+// If the 'zenoss' section in the configuration has the option: global = true
+// then all alerts are sent to Zenoss without the need to explicitly state it
+// in the TICKscript.
+//
+// Example:
+//    [zenoss]
+//      enabled = true
+//      url = "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router"
+//      global = true
+//      state-changes-only = true
+//
+// Example:
+//    stream
+//         |alert()
+//             .zenoss()
+//                 .summary('Alert {{ .ID }}')
+//                 .message('{{ .Message }}')
+//                 .eventClass('/App')
+//
+// Send alert to Zenoss using default url.
+// tick:property
+func (n *AlertNodeData) Zenoss() *ZenossHandler {
+	zenoss := &ZenossHandler{
+		AlertNodeData: n,
+		Summary:       "{{ .Message }}", // default mapping
+	}
+	n.ZenossHandlers = append(n.ZenossHandlers, zenoss)
+	return zenoss
+}
+
+// tick:embedded:AlertNode.Zenoss
+type ZenossHandler struct {
+	*AlertNodeData `json:"-"`
+
+	// Zenoss API URL to post alerts.
+	// If empty uses the URL from the configuration.
+	URL string `json:"url"`
+
+	// Username for BASIC authentication.
+	// If empty uses username from the configuration.
+	Username string `json:"username"`
+
+	// Password for BASIC authentication.
+	// If empty uses password from the configuration.
+	Password string `json:"password"`
+
+	// Action (router name).
+	// If empty uses value from the configuration.
+	Action string `json:"action"`
+
+	// Router method.
+	// If empty uses value from the configuration.
+	Method string `json:"method"`
+
+	// Event type.
+	// If empty uses value from the configuration.
+	Type string `json:"type"`
+
+	// Temporary transaction ID.
+	// If empty uses value from the configuration.
+	TID int64 `json:"tid"`
+
+	// Summary of the event.
+	Summary string `json:"summary"`
+
+	// Device (typically IP or hostname).
+	Device string `json:"device"`
+
+	// Component related to the event.
+	Component string `json:"component"`
+
+	// Event class key.
+	EventClassKey string `json:"evclasskey"`
+
+	// Event class.
+	EventClass string `json:"evclass"`
+
+	// Message related to the event.
+	Message string `json:"message"`
+
+	// Collector.
+	Collector string `json:"collector"`
+
+	// Custom fields.
+	// tick:ignore
+	CustomFieldsMap map[string]interface{} `tick:"CustomField" json:"customField"`
+}
+
+// CustomField adds a field to event data.
+// tick:property
+func (s *ZenossHandler) CustomField(key string, value interface{}) *ZenossHandler {
+	if s.CustomFieldsMap == nil {
+		s.CustomFieldsMap = make(map[string]interface{})
+	}
+	s.CustomFieldsMap[key] = value
+	return s
+}
diff --git a/pipeline/tick/alert.go b/pipeline/tick/alert.go
index 6f4948f5..01bbb04f 100644
--- a/pipeline/tick/alert.go
+++ b/pipeline/tick/alert.go
@@ -210,6 +210,8 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 		n.Dot("kafka").
 			Dot("cluster", h.Cluster).
 			Dot("kafkaTopic", h.KafkaTopic).
+			DotIf("disablePartitionById", h.IsDisablePartitionById).
+			Dot("partitionHashAlgorithm", h.PartitionHashAlgorithm).
 			Dot("template", h.Template)
 	}
 
@@ -256,5 +258,31 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 		}
 	}
 
+	for _, h := range a.ZenossHandlers {
+		n.Dot("zenoss").
+			Dot("action", h.Action).
+			Dot("method", h.Method).
+			Dot("type", h.Type).
+			Dot("TID", h.TID).
+			// standard event data element fields
+			Dot("summary", h.Summary).
+			Dot("device", h.Device).
+			Dot("component", h.Component).
+			Dot("eventClassKey", h.EventClassKey).
+			Dot("eventClass", h.EventClass).
+			Dot("collector", h.Collector).
+			Dot("message", h.Message)
+
+		// Use stable key order
+		keys := make([]string, 0, len(h.CustomFieldsMap))
+		for k := range h.CustomFieldsMap {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			n.Dot("customField", k, h.CustomFieldsMap[k])
+		}
+	}
+
 	return n.prev, n.err
 }
diff --git a/server/config.go b/server/config.go
index 336bbb3c..7a3bc574 100644
--- a/server/config.go
+++ b/server/config.go
@@ -11,9 +11,14 @@ import (
 	"strings"
 	"time"
 
+	"github.com/influxdata/influxdb/services/collectd"
+	"github.com/influxdata/influxdb/services/graphite"
+	"github.com/influxdata/influxdb/services/opentsdb"
 	"github.com/influxdata/kapacitor/command"
 	"github.com/influxdata/kapacitor/services/alert"
 	"github.com/influxdata/kapacitor/services/alerta"
+	"github.com/influxdata/kapacitor/services/auth"
+	authservice "github.com/influxdata/kapacitor/services/auth"
 	"github.com/influxdata/kapacitor/services/azure"
 	"github.com/influxdata/kapacitor/services/bigpanda"
 	"github.com/influxdata/kapacitor/services/config"
@@ -61,17 +66,15 @@ import (
 	"github.com/influxdata/kapacitor/services/udf"
 	"github.com/influxdata/kapacitor/services/udp"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/influxdata/kapacitor/tlsconfig"
 	"github.com/pkg/errors"
-
-	"github.com/influxdata/influxdb/services/collectd"
-	"github.com/influxdata/influxdb/services/graphite"
-	"github.com/influxdata/influxdb/services/opentsdb"
 )
 
 // Config represents the configuration format for the kapacitord binary.
 type Config struct {
 	Alert          alert.Config      `toml:"alert"`
+	Auth           auth.Config       `toml:"auth"`
 	HTTP           httpd.Config      `toml:"http"`
 	Replay         replay.Config     `toml:"replay"`
 	Storage        storage.Config    `toml:"storage"`
@@ -110,6 +113,7 @@ type Config struct {
 	Teams      teams.Config      `toml:"teams" override:"teams"`
 	Telegram   telegram.Config   `toml:"telegram" override:"telegram"`
 	VictorOps  victorops.Config  `toml:"victorops" override:"victorops"`
+	Zenoss     zenoss.Config     `toml:"zenoss" override:"zenoss"`
 
 	// Discovery for scraping
 	Scraper         []scraper.Config          `toml:"scraper" override:"scraper,element-key=name"`
@@ -150,6 +154,7 @@ func NewConfig() *Config {
 	}
 
 	c.Alert = alert.NewConfig()
+	c.Auth = authservice.NewDisabledConfig()
 	c.HTTP = httpd.NewConfig()
 	c.Storage = storage.NewConfig()
 	c.Replay = replay.NewConfig()
@@ -183,6 +188,7 @@ func NewConfig() *Config {
 	c.SNMPTrap = snmptrap.NewConfig()
 	c.Telegram = telegram.NewConfig()
 	c.VictorOps = victorops.NewConfig()
+	c.Zenoss = zenoss.NewConfig()
 
 	c.Reporting = reporting.NewConfig()
 	c.Stats = stats.NewConfig()
@@ -228,6 +234,9 @@ func (c *Config) Validate() error {
 	if err := c.Replay.Validate(); err != nil {
 		return errors.Wrap(err, "replay")
 	}
+	if err := c.Auth.Validate(); err != nil {
+		return errors.Wrap(err, "auth")
+	}
 	if err := c.Storage.Validate(); err != nil {
 		return errors.Wrap(err, "storage")
 	}
@@ -344,6 +353,9 @@ func (c *Config) Validate() error {
 	if err := c.VictorOps.Validate(); err != nil {
 		return errors.Wrap(err, "victorops")
 	}
+	if err := c.Zenoss.Validate(); err != nil {
+		return errors.Wrap(err, "zenoss")
+	}
 
 	if err := c.UDF.Validate(); err != nil {
 		return errors.Wrap(err, "udf")
diff --git a/server/server.go b/server/server.go
index 82194039..8833fd30 100644
--- a/server/server.go
+++ b/server/server.go
@@ -24,6 +24,7 @@ import (
 	"github.com/influxdata/kapacitor/server/vars"
 	"github.com/influxdata/kapacitor/services/alert"
 	"github.com/influxdata/kapacitor/services/alerta"
+	authservice "github.com/influxdata/kapacitor/services/auth"
 	"github.com/influxdata/kapacitor/services/azure"
 	"github.com/influxdata/kapacitor/services/bigpanda"
 	"github.com/influxdata/kapacitor/services/config"
@@ -74,6 +75,7 @@ import (
 	"github.com/influxdata/kapacitor/services/udf"
 	"github.com/influxdata/kapacitor/services/udp"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/influxdata/kapacitor/uuid"
 	"github.com/influxdata/kapacitor/waiter"
 	"github.com/pkg/errors"
@@ -222,7 +224,15 @@ func New(c *Config, buildInfo BuildInfo, diagService *diagnostic.Service) (*Serv
 	// Append Kapacitor services.
 	s.initHTTPDService()
 	s.appendStorageService()
-	s.appendAuthService()
+
+	if c.Auth.Enabled {
+		if err := s.appendEnabledAuthService(); err != nil {
+			return nil, err
+		}
+	} else {
+		s.appendNoAuthService()
+	}
+
 	s.appendConfigOverrideService()
 	s.appendTesterService()
 	s.appendSideloadService()
@@ -274,6 +284,7 @@ func New(c *Config, buildInfo BuildInfo, diagService *diagnostic.Service) (*Serv
 	s.appendSensuService()
 	s.appendTalkService()
 	s.appendVictorOpsService()
+	s.appendZenossService()
 
 	// Append alert service
 	s.appendAlertService()
@@ -578,7 +589,7 @@ func (s *Server) appendUDFService() {
 	s.AppendService("udf", srv)
 }
 
-func (s *Server) appendAuthService() {
+func (s *Server) appendNoAuthService() {
 	d := s.DiagService.NewNoAuthHandler()
 	srv := noauth.NewService(d)
 
@@ -587,6 +598,23 @@ func (s *Server) appendAuthService() {
 	s.AppendService("auth", srv)
 }
 
+func (s *Server) appendEnabledAuthService() error {
+	d := s.DiagService.NewAuthHandler()
+
+	srv, err := authservice.NewService(s.config.Auth, d)
+	if err != nil {
+		return err
+	}
+	srv.HTTPDService = s.HTTPDService
+	srv.StorageService = s.StorageService
+
+	s.AuthService = srv
+	s.HTTPDService.Handler.AuthService = srv
+
+	s.AppendService("auth", srv)
+	return nil
+}
+
 func (s *Server) appendMQTTService() error {
 	cs := s.config.MQTT
 	d := s.DiagService.NewMQTTHandler()
@@ -1035,6 +1063,18 @@ func (s *Server) appendServiceNowService() {
 	s.AppendService("servicenow", srv)
 }
 
+func (s *Server) appendZenossService() {
+	c := s.config.Zenoss
+	d := s.DiagService.NewZenossHandler()
+	srv := zenoss.NewService(c, d)
+
+	s.TaskMaster.ZenossService = srv
+	s.AlertService.ZenossService = srv
+
+	s.SetDynamicService("zenoss", srv)
+	s.AppendService("zenoss", srv)
+}
+
 // Err returns an error channel that multiplexes all out of band errors received from all services.
 func (s *Server) Err() <-chan error { return s.err }
 
diff --git a/services/alert/dao_easyjson.go b/services/alert/dao_easyjson.go
index 9b13b26f..e4399555 100644
--- a/services/alert/dao_easyjson.go
+++ b/services/alert/dao_easyjson.go
@@ -4,10 +4,11 @@ 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/alert/service.go b/services/alert/service.go
index 4b952e37..749dfbd8 100644
--- a/services/alert/service.go
+++ b/services/alert/service.go
@@ -35,6 +35,7 @@ import (
 	"github.com/influxdata/kapacitor/services/teams"
 	"github.com/influxdata/kapacitor/services/telegram"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 )
@@ -148,6 +149,9 @@ type Service struct {
 	ServiceNowService interface {
 		Handler(servicenow.HandlerConfig, ...keyvalue.T) alert.Handler
 	}
+	ZenossService interface {
+		Handler(zenoss.HandlerConfig, ...keyvalue.T) alert.Handler
+	}
 }
 
 func NewService(d Diagnostic) *Service {
@@ -1014,6 +1018,14 @@ func (s *Service) createHandlerFromSpec(spec HandlerSpec) (handler, error) {
 		}
 		h = s.VictorOpsService.Handler(c, ctx...)
 		h = newExternalHandler(h)
+	case "zenoss":
+		c := zenoss.HandlerConfig{}
+		err = decodeOptions(spec.Options, &c)
+		if err != nil {
+			return handler{}, err
+		}
+		h = s.ZenossService.Handler(c, ctx...)
+		h = newExternalHandler(h)
 	default:
 		err = fmt.Errorf("unsupported action kind %q", spec.Kind)
 	}
diff --git a/services/auth/cache.go b/services/auth/cache.go
new file mode 100644
index 00000000..e4871052
--- /dev/null
+++ b/services/auth/cache.go
@@ -0,0 +1,72 @@
+package auth
+
+import (
+	"sync"
+	"time"
+
+	"github.com/influxdata/kapacitor/auth"
+)
+
+type UserCache interface {
+	Get(username string) (auth.User, bool)
+	Set(auth.User)
+	Delete(username string)
+	DeleteAll()
+}
+
+// In memory based implementation of User cache
+// Lazily expires items from the cache.
+// All methods are goroutine safe.
+type memUserCache struct {
+	mu         sync.RWMutex
+	cache      map[string]cacheItem
+	expiration time.Duration
+}
+
+type cacheItem struct {
+	User       auth.User
+	Expiration time.Time
+}
+
+func newMemUserCache(expiration time.Duration) *memUserCache {
+	return &memUserCache{
+		cache:      make(map[string]cacheItem),
+		expiration: expiration,
+	}
+}
+
+func (c *memUserCache) Get(username string) (auth.User, bool) {
+	c.mu.RLock()
+	item, ok := c.cache[username]
+	c.mu.RUnlock()
+	if item.Expiration.Before(time.Now()) {
+		c.mu.Lock()
+		delete(c.cache, username)
+		c.mu.Unlock()
+		ok = false
+	}
+	return item.User, ok
+}
+
+func (c *memUserCache) Set(u auth.User) {
+	c.mu.Lock()
+	c.cache[u.Name()] = cacheItem{
+		User:       u,
+		Expiration: time.Now().Add(c.expiration),
+	}
+	c.mu.Unlock()
+}
+
+func (c *memUserCache) Delete(username string) {
+	c.mu.Lock()
+	delete(c.cache, username)
+	c.mu.Unlock()
+}
+
+func (c *memUserCache) DeleteAll() {
+	c.mu.Lock()
+	for username := range c.cache {
+		delete(c.cache, username)
+	}
+	c.mu.Unlock()
+}
diff --git a/services/auth/config.go b/services/auth/config.go
new file mode 100644
index 00000000..b16a6b68
--- /dev/null
+++ b/services/auth/config.go
@@ -0,0 +1,53 @@
+package auth
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/influxdata/influxdb/toml"
+	"golang.org/x/crypto/bcrypt"
+)
+
+const (
+	// Default cost is 10
+	DefaultBcryptCost      = bcrypt.DefaultCost
+	DefaultCacheExpiration = 10 * time.Minute
+)
+
+type Config struct {
+	Enabled                bool          `toml:"enabled"`
+	CacheExpiration        toml.Duration `toml:"cache-expiration"`
+	BcryptCost             int           `toml:"bcrypt-cost"`
+	MetaAddr               string        `toml:"meta-addr"`
+	MetaUsername           string        `toml:"meta-username"`
+	MetaPassword           string        `toml:"meta-password"`
+	MetaUseTLS             bool          `toml:"meta-use-tls"`
+	MetaCA                 string        `toml:"meta-ca"`
+	MetaCert               string        `toml:"meta-cert"`
+	MetaKey                string        `toml:"meta-key"`
+	MetaInsecureSkipVerify bool          `toml:"meta-insecure-skip-verify"`
+}
+
+func NewDisabledConfig() Config {
+	return Config{
+		Enabled: false,
+	}
+}
+
+func NewEnabledConfig() Config {
+	return Config{
+		Enabled:         true,
+		CacheExpiration: toml.Duration(DefaultCacheExpiration),
+		BcryptCost:      DefaultBcryptCost,
+	}
+}
+
+func (c Config) Validate() error {
+	if !c.Enabled {
+		return nil
+	}
+	if c.BcryptCost < bcrypt.MinCost {
+		return fmt.Errorf("must provide a bcrypt cost >= %d, got %d", bcrypt.MinCost, c.BcryptCost)
+	}
+	return nil
+}
diff --git a/services/auth/dao.go b/services/auth/dao.go
new file mode 100644
index 00000000..adbcbabc
--- /dev/null
+++ b/services/auth/dao.go
@@ -0,0 +1,163 @@
+package auth
+
+import (
+	"bytes"
+	"encoding/gob"
+	"errors"
+
+	"github.com/influxdata/kapacitor/services/storage"
+)
+
+var (
+	ErrUserExists   = errors.New("user already exists")
+	ErrNoUserExists = errors.New("no user exists")
+)
+
+type UserDAO interface {
+	// Retrieve a user
+	Get(username string) (User, error)
+
+	// Create a user.
+	// ErrUserExists is returned if a user already exists with the same username.
+	Create(u User) error
+
+	// Replace an existing user.
+	// ErrNoUserExists is returned if the user does not exist.
+	Replace(u User) error
+
+	// Delete a user.
+	// It is not an error to delete an non-existent user.
+	Delete(username string) error
+
+	// List users matching a pattern on username.
+	// The pattern is shell/glob matching see https://golang.org/pkg/path/#Match
+	// Offset and limit are pagination bounds. Offset is inclusive starting at index 0.
+	// More results may exist while the number of returned items is equal to limit.
+	List(pattern string, offset, limit int) ([]User, error)
+}
+
+//--------------------------------------------------------------------
+// The following structures are stored in a database via gob encoding.
+// Changes to the structures could break existing data.
+//
+// Many of these structures are exact copies of structures found elsewhere,
+// this is intentional so that all structures stored in the database are
+// defined here and nowhere else. So as to not accidentally change
+// the gob serialization format in incompatible ways.
+
+type Privilege int
+
+const (
+	NoPrivileges Privilege = iota
+
+	ReadPrivilege
+	WritePrivilege
+	DeletePrivilege
+
+	AllPrivileges
+)
+
+// A user with is set of permissions
+type User struct {
+	Name       string
+	Admin      bool
+	Hash       []byte
+	Privileges map[string][]Privilege
+}
+
+type rawUser User
+
+func (u User) ObjectID() string {
+	return u.Name
+}
+
+func (u User) MarshalBinary() ([]byte, error) {
+	var buf bytes.Buffer
+	enc := gob.NewEncoder(&buf)
+	err := enc.Encode(rawUser(u))
+	return buf.Bytes(), err
+}
+
+func (u *User) UnmarshalBinary(data []byte) error {
+	dec := gob.NewDecoder(bytes.NewReader(data))
+	return dec.Decode((*rawUser)(u))
+}
+
+const (
+	// Name of username index
+	usernameIndex = "username"
+)
+
+// Key/Value store based implementation of the UserDAO
+type userKV struct {
+	store *storage.IndexedStore
+}
+
+func newUserKV(store storage.Interface) (*userKV, error) {
+	c := storage.DefaultIndexedStoreConfig("users", func() storage.BinaryObject {
+		return new(User)
+	})
+	c.Indexes = []storage.Index{{
+		Name:   usernameIndex,
+		Unique: true,
+		ValueFunc: func(o storage.BinaryObject) (string, error) {
+			return o.ObjectID(), nil
+		},
+	}}
+	istore, err := storage.NewIndexedStore(store, c)
+	if err != nil {
+		return nil, err
+	}
+	return &userKV{
+		store: istore,
+	}, nil
+}
+
+func (kv *userKV) error(err error) error {
+	if err == storage.ErrNoObjectExists {
+		return ErrNoUserExists
+	} else if err == storage.ErrObjectExists {
+		return ErrUserExists
+	}
+	return err
+}
+
+func (kv *userKV) Get(username string) (User, error) {
+	o, err := kv.store.Get(username)
+	if err != nil {
+		return User{}, kv.error(err)
+	}
+	u, ok := o.(*User)
+	if !ok {
+		return User{}, storage.ImpossibleTypeErr(u, o)
+	}
+	return *u, nil
+}
+
+func (kv *userKV) Create(u User) error {
+	return kv.error(kv.store.Create(&u))
+}
+
+func (kv *userKV) Replace(u User) error {
+	return kv.error(kv.store.Replace(&u))
+}
+
+func (kv *userKV) Delete(username string) error {
+	return kv.error(kv.store.Delete(username))
+}
+
+func (kv *userKV) List(pattern string, offset, limit int) ([]User, error) {
+	objects, err := kv.store.List(usernameIndex, pattern, offset, limit)
+	if err != nil {
+		return nil, kv.error(err)
+	}
+	users := make([]User, len(objects))
+	for i, o := range objects {
+		u, ok := o.(*User)
+		if !ok {
+			return nil, storage.ImpossibleTypeErr(u, o)
+		}
+		users[i] = *u
+	}
+	return users, nil
+}
diff --git a/services/auth/meta/client.go b/services/auth/meta/client.go
new file mode 100644
index 00000000..f83dae1b
--- /dev/null
+++ b/services/auth/meta/client.go
@@ -0,0 +1,398 @@
+// Package meta provides a client to the meta node to allow Kapacitor to perform auth.
+// The real home of the meta client is in the InfluxDB Enterprise repo, however kapacitor's
+// current dependency in influxdb is stuck on 1.1.
+// This package allows kapacitor to use the newest meta client API but avoid having to
+// upgrade kapacitor's dependency on influxdb which is now a couple years behind.
+package meta
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
+)
+
+const controlClientUA = "InfluxDB Cluster Client"
+
+const (
+	// DefaultTimeout is the duration that the client will continue to
+	// retry a failed operation.
+	DefaultTimeout = 10 * time.Second
+
+	// DefaultBackoffCap is the maximum duration a backoff value can
+	// have.
+	DefaultBackoffCap = DefaultTimeout / 3
+
+	// MaxClientRedirects is the maximum number of redirects the Client will
+	// attempt to follow for a single request, before failing.
+	MaxClientRedirects = 5
+)
+
+// AuthType identifies a method of authentication.
+type AuthType int
+
+const (
+	// NoAuth means no authentication will be used.
+	NoAuth AuthType = iota
+	// BasicAuth means basic user authentication will be used.
+	BasicAuth
+	// BearerAuth means JWT tokens will be used.
+	BearerAuth
+)
+
+// Client is a client for the HTTP API exposed by a Plutonium Meta node.
+type Client struct {
+	client *http.Client
+
+	metaBind string      // Address to bind to meta node.
+	useTLS   bool        // Whether to use TLS protocol.
+	tls      *tls.Config // TLS configuration to use if not skipping verification
+	skipTLS  bool        // Whether to skip TLS verification
+
+	timeout    time.Duration // Duration to timeout requests to meta node.
+	backoffCap time.Duration // Maximum wait between request retries.
+
+	authType AuthType
+	username string // Name of user if authentication is enabled.
+	password string // User's password if authentication is enabled.
+	secret   string // Shared secret for JWT token signing.
+}
+
+// ClientOption is a functional option for a Client.
+type ClientOption func(c *Client)
+
+// UseAuth sets the authentication type and credentials that will be used.
+var UseAuth = func(t AuthType, username, password, secret string) ClientOption {
+	return func(c *Client) {
+		c.authType = t
+		c.username = username
+		c.password = password
+		c.secret = secret
+	}
+}
+
+// WithTLS specifies if communication to the meta node uses the specified TLS config if any
+// and if we skip verifying certificates, allowing for self-signed certificates.
+var WithTLS = func(tlsConfig *tls.Config, useTLS bool, skip bool) ClientOption {
+	return func(c *Client) {
+		c.useTLS = useTLS
+		if skip {
+			c.skipTLS = true
+			c.tls = &tls.Config{
+				InsecureSkipVerify: true,
+			}
+		} else if useTLS {
+			c.tls = tlsConfig
+		}
+
+		if t, ok := c.client.Transport.(*http.Transport); ok {
+			t.TLSClientConfig = c.tls
+		} else {
+			c.client.Transport = &http.Transport{
+				TLSClientConfig: c.tls,
+			}
+		}
+	}
+}
+
+// WithTimeout specifies the duration the client will continue to retry
+// failed operations before giving up.
+var WithTimeout = func(d time.Duration) ClientOption {
+	return func(c *Client) {
+		c.timeout = d
+		c.backoffCap = c.timeout / 3
+		if t, ok := c.client.Transport.(*http.Transport); ok {
+			t.ResponseHeaderTimeout = d
+		} else {
+			c.client.Transport = &http.Transport{
+				ResponseHeaderTimeout: d,
+			}
+		}
+	}
+}
+
+// NewClient returns a new Client, which will make requests to the Meta
+// node listening on addr. New accepts zero or more functional options
+// for configuring aspects of the returned Client.
+//
+// By default, a Client does not send communications over a TLS
+// connection. TLS can be enforced using the WithTLS option. Output and
+// errors are logged to stdout and stderr, respectively; use the
+// appropriate options to set those outputs.
+func NewClient(addr string, options ...ClientOption) *Client {
+	c := &Client{
+		metaBind:   addr,
+		timeout:    DefaultTimeout,
+		backoffCap: DefaultBackoffCap,
+	}
+
+	c.setRedirectClient()
+
+	for _, option := range options {
+		option(c)
+	}
+
+	return c
+}
+
+// setRedirectClient initializes the HTTP client
+// to set Auth credentials on a redirect if needed.
+func (c *Client) setRedirectClient() {
+	c.client = &http.Client{}
+
+	// Configure HTTP client to forward request headers on redirect.
+	c.client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
+		if len(via) >= 10 {
+			return errors.New("too many redirects")
+		} else if len(via) == 0 {
+			return nil
+		}
+		for attr, val := range via[0].Header {
+			if _, ok := req.Header[attr]; !ok {
+				req.Header[attr] = val
+			}
+		}
+		return c.setAuth(req)
+	}
+}
+
+// setAuth adds the correct Authorization header to a request, if needed
+func (c *Client) setAuth(r *http.Request) error {
+	switch c.authType {
+	case BasicAuth:
+		r.SetBasicAuth(c.username, c.password)
+	case BearerAuth:
+		token := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
+			"username": c.username,
+			"exp":      time.Now().Add(5 * time.Minute).Unix(),
+		})
+		tokenStr, err := token.SignedString([]byte(c.secret))
+		if err != nil {
+			return err
+		}
+		r.Header.Set("Authorization", fmt.Sprintf("Bearer %s", tokenStr))
+	}
+
+	return nil
+}
+
+// retryWithBackoff will continue to retry remote operation f until the
+// the success predicate is satisfied. On failure, backoff will be
+// exponentially increased up to a maximum capacity.
+func (c *Client) retryWithBackoff(f func() error) error {
+	var (
+		base    = 10.0 // Will end up being 10 milliseconds.
+		timeout = time.NewTimer(c.timeout)
+		try     = time.NewTimer(0)
+		retries float64
+		err     error
+	)
+	defer timeout.Stop()
+	defer try.Stop()
+
+	for {
+		select {
+		case <-timeout.C:
+			return ErrTimeout(err)
+		case <-try.C:
+			if err = f(); err == nil {
+				return nil
+			} else if _, ok := err.(FatalError); ok {
+				// A fatal error has occurred; don't retry.
+				return err
+			}
+
+			// On non-fatal failure exponentially increase backoff up to
+			// cap.
+			retries++
+			d := time.Duration(math.Pow(2.0*base, retries)) * time.Millisecond
+			if d > c.backoffCap {
+				d = c.backoffCap
+			}
+			try.Reset(d)
+			// TODO(cwolff): we should log an error here, like the original code did.
+		}
+	}
+}
+
+// do carries out an HTTP request. If host is empty it will use the host
+// associated with the Client.
+//
+// do has the abilty to automatially redirect a request if the initial
+// response status code is 307 (temporary redirect). It will follow the
+// Location header and replay the same request including the body.
+func (c *Client) do(method, host, path string, query url.Values, ct string, body io.Reader) (*http.Response, error) {
+	var (
+		redirects int
+		buf       bytes.Buffer
+	)
+
+	if host == "" {
+		host = c.metaBind
+	}
+
+	// We need to save the body in case we have to redirect.
+	if body != nil {
+		if _, err := io.Copy(&buf, body); err != nil {
+			return nil, err
+		}
+	}
+
+	u := &url.URL{}
+	u.Path = path
+	u.RawQuery = query.Encode()
+	u.Host = host
+	u.Scheme = "http"
+	if c.useTLS {
+		u.Scheme = "https"
+	}
+	u.RawQuery = query.Encode()
+	u.RequestURI()
+
+	location := u.String()
+
+	for {
+		// Set the body to a copy of the original body passed in.
+		// This allows us to reuse it each time.
+		if body != nil {
+			body = ioutil.NopCloser(bytes.NewReader(buf.Bytes()))
+		}
+
+		req, err := http.NewRequest(method, location, body)
+		if err != nil {
+			return nil, err
+		}
+		req.Header.Add("Accept", "application/json")
+
+		if err = c.setAuth(req); err != nil {
+			return nil, err
+		}
+
+		req.Header.Set("User-Agent", controlClientUA)
+		if ct != "" {
+			req.Header.Set("Content-Type", ct)
+		}
+
+		resp, err := c.client.Do(req)
+		if err != nil {
+			return resp, err
+		}
+
+		if resp.StatusCode == http.StatusTemporaryRedirect {
+			if redirects > MaxClientRedirects {
+				return resp, ErrMaximumRedirectsReached
+			}
+
+			// We're done with the current response so make sure we
+			// close the body.
+			resp.Body.Close()
+
+			redirects++
+			location = resp.Header.Get("Location")
+			continue
+		}
+
+		// 307 status code *not* returned, so we're done.
+		return resp, err
+	}
+}
+
+// get is a helper method for making a GET request to the node the
+// client is bound to, using do. get sets application/json as the
+// Content-Type.
+func (c *Client) get(path string, query url.Values) (*http.Response, error) {
+	return c.do("GET", c.metaBind, path, query, "application/json", nil)
+}
+
+// errorFromResponse will read r and attempt to unmarshal JSON from r
+// into an Error. If r is empty, then an Error without any message will
+// be returned.
+// If r is not empty, but it's not possible to unmarshal r into a valid
+// Error then an error will be returned.
+func errorFromResponse(r io.Reader, code int) (Error, error) {
+	var errMsg Error
+	if err := json.NewDecoder(r).Decode(&errMsg); err != nil {
+		if err != io.EOF {
+			return Error{}, err
+		}
+		// In this case r was empty.
+	}
+	errMsg.Code = code
+	return errMsg, nil
+}
+
+// Users returns a given user or a list of all users if name is empty.
+// If the user is not found an empty list is returned without an error.
+func (c *Client) Users(name string) ([]User, error) {
+	var users struct {
+		Users []User `json:"users"`
+	}
+
+	data := url.Values{"name": {name}}
+
+	// cmd will be retried with back-off, until success or timeout.
+	cmd := func() error {
+		resp, err := c.get("/user", data)
+		if err != nil {
+			return err
+		}
+		if resp == nil {
+			return fmt.Errorf("unable to get user: %q", name)
+		}
+
+		defer resp.Body.Close()
+
+		switch resp.StatusCode {
+		case http.StatusOK:
+			return json.NewDecoder(resp.Body).Decode(&users)
+		default:
+			msg, err := errorFromResponse(resp.Body, resp.StatusCode)
+			if err != nil {
+				return err
+			}
+			return msg
+		}
+	}
+
+	// Run cmd and wait for timeout error, if any.
+	if err := c.retryWithBackoff(cmd); err != nil {
+		return nil, err
+	}
+
+	return users.Users, nil
+}
+
+func (c *Client) CheckPass(name, pass string) error {
+	data := url.Values{
+		"name":       {name},
+		"password":   {pass},
+		"permission": {"NoPermissions"}, // because we just want to check the password here, not the other stuff.
+		"resource":   {"_"},
+	}
+	cmd := func() error {
+		resp, err := c.get("/authorized", data)
+		if err != nil {
+			return err
+		}
+		switch resp.StatusCode {
+		case http.StatusOK:
+			return nil
+		default:
+			msg, err := errorFromResponse(resp.Body, resp.StatusCode)
+			if err != nil {
+				return err
+			}
+			return msg
+		}
+	}
+	return c.retryWithBackoff(cmd)
+}
diff --git a/services/auth/meta/errors.go b/services/auth/meta/errors.go
new file mode 100644
index 00000000..df93041f
--- /dev/null
+++ b/services/auth/meta/errors.go
@@ -0,0 +1,78 @@
+package meta
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	// ErrAddressMissing is returned when a required addr parameter is
+	// missing.
+	ErrAddressMissing = errors.New("addr value is empty")
+
+	// ErrHTTPAddressMissing is returned when a required httpAddr
+	// parameter is missing.
+	ErrHTTPAddressMissing = errors.New("httpAddr value is empty")
+
+	// ErrTCPAddressMissing is returned when a required tcpAddr
+	// parameter is missing.
+	ErrTCPAddressMissing = errors.New("tcpAddr value is empty")
+
+	// ErrLeaseNameMissing is returned when a required lease name is
+	// missing.
+	ErrLeaseNameMissing = errors.New("lease name is empty")
+
+	// ErrLeaseConflict is returned when there is a conflict while
+	// attempting to require a lease.
+	ErrLeaseConflict = errors.New("another node owns the lease")
+
+	// ErrMaximumRedirectsReached is returned when a Client has been
+	// redirected too many times for a single request.
+	ErrMaximumRedirectsReached = errors.New("too many redirects")
+
+	// ErrNotMetaNode is returned when attempting to add a non-meta node
+	// as a meta node in the cluster.
+	ErrNotMetaNode = errors.New("not a meta node")
+
+	// ErrEmptyCluster is returned when there are no nodes known to the
+	// cluster.
+	ErrEmptyCluster = errors.New("cluster is empty")
+
+	ErrEmptyStatus = errors.New("status is empty")
+
+	// ErrAntiEntropyDisabled is returned when the anti-entropy service
+	// returns a 501 Not Implemented HTTP status.
+	ErrAntiEntropyDisabled = errors.New("anti-entropy service is not enabled")
+)
+
+// A TimeoutError is returned when an operation on the meta node API was
+// not successful, and repeated attempts to succeed have failed.
+//
+// TimeoutError makes the last underlying error available in the
+// Context field.
+type TimeoutError struct {
+	Context error
+}
+
+func (e TimeoutError) Error() string {
+	return fmt.Sprintf("operation timed out with error: %v", e.Context)
+}
+
+// ErrTimeout returns a new TimeoutError.
+func ErrTimeout(ctx error) TimeoutError {
+	return TimeoutError{Context: ctx}
+}
+
+// A FatalError is returned when an operation on the meta node API was
+// not successful and retrying the operation would always result in the
+// same error being returned.
+//
+// FatalError makes the last underlying error available in the
+// Context field.
+type FatalError struct {
+	Context error
+}
+
+func (e FatalError) Error() string {
+	return fmt.Sprintf("operation exited with error: %v", e.Context)
+}
diff --git a/services/auth/meta/types.go b/services/auth/meta/types.go
new file mode 100644
index 00000000..2bd36352
--- /dev/null
+++ b/services/auth/meta/types.go
@@ -0,0 +1,148 @@
+package meta
+
+import (
+	"fmt"
+)
+
+// Error represent an error message sent in a response body from a meta
+// node.
+type Error struct {
+	Message string `json:"error"`
+	Code    int    `json:"-"`
+}
+
+// NewError returns a new Error value.
+func NewError(msg string, code int) Error {
+	return Error{Message: msg, Code: code}
+}
+
+func (e Error) Error() string {
+	return e.Message
+}
+
+// User holds information about a user's credentials and permissions.
+type User struct {
+	Name        string                  `json:"name"`
+	Hash        string                  `json:"hash"`
+	Permissions map[string][]Permission `json:"permissions,omitempty"`
+}
+
+// Permission type
+// NOTE: Keep this type up-to-date with the corresponding type in meta/data.go
+// The type cannot be imported as it creates an import cycle.
+type Permission int
+
+const (
+	NoPermissions                   Permission = 0
+	ViewAdminPermission             Permission = 1
+	ViewChronografPermission        Permission = 2
+	CreateDatabasePermission        Permission = 3
+	CreateUserAndRolePermission     Permission = 4
+	AddRemoveNodePermission         Permission = 5
+	DropDatabasePermission          Permission = 6
+	DropDataPermission              Permission = 7
+	ReadDataPermission              Permission = 8
+	WriteDataPermission             Permission = 9
+	RebalancePermission             Permission = 10
+	ManageShardPermission           Permission = 11
+	ManageContinuousQueryPermission Permission = 12
+	ManageQueryPermission           Permission = 13
+	ManageSubscriptionPermission    Permission = 14
+	MonitorPermission               Permission = 15
+	CopyShardPermission             Permission = 16
+	KapacitorAPIPermission          Permission = 17
+	KapacitorConfigAPIPermission    Permission = 18
+)
+
+// MarshalText converts a Permission into a textual represenation.
+func (p Permission) MarshalText() ([]byte, error) {
+	switch p {
+	case NoPermissions:
+		return []byte("NoPermissions"), nil
+	case ViewAdminPermission:
+		return []byte("ViewAdmin"), nil
+	case ViewChronografPermission:
+		return []byte("ViewChronograf"), nil
+	case CreateDatabasePermission:
+		return []byte("CreateDatabase"), nil
+	case CreateUserAndRolePermission:
+		return []byte("CreateUserAndRole"), nil
+	case AddRemoveNodePermission:
+		return []byte("AddRemoveNode"), nil
+	case DropDatabasePermission:
+		return []byte("DropDatabase"), nil
+	case DropDataPermission:
+		return []byte("DropData"), nil
+	case ReadDataPermission:
+		return []byte("ReadData"), nil
+	case WriteDataPermission:
+		return []byte("WriteData"), nil
+	case RebalancePermission:
+		return []byte("Rebalance"), nil
+	case ManageShardPermission:
+		return []byte("ManageShard"), nil
+	case ManageContinuousQueryPermission:
+		return []byte("ManageContinuousQuery"), nil
+	case ManageQueryPermission:
+		return []byte("ManageQuery"), nil
+	case ManageSubscriptionPermission:
+		return []byte("ManageSubscription"), nil
+	case MonitorPermission:
+		return []byte("Monitor"), nil
+	case CopyShardPermission:
+		return []byte("CopyShard"), nil
+	case KapacitorAPIPermission:
+		return []byte("KapacitorAPI"), nil
+	case KapacitorConfigAPIPermission:
+		return []byte("KapacitorConfigAPI"), nil
+	default:
+		return nil, fmt.Errorf("invalid permission: %d", p)
+	}
+}
+
+// UnmarshalText sets the permission based on the textual represenation.
+func (p *Permission) UnmarshalText(t []byte) error {
+	switch s := string(t); s {
+	case "NoPermissions":
+		*p = NoPermissions
+	case "ViewAdmin":
+		*p = ViewAdminPermission
+	case "ViewChronograf":
+		*p = ViewChronografPermission
+	case "CreateDatabase":
+		*p = CreateDatabasePermission
+	case "CreateUserAndRole":
+		*p = CreateUserAndRolePermission
+	case "AddRemoveNode":
+		*p = AddRemoveNodePermission
+	case "DropDatabase":
+		*p = DropDatabasePermission
+	case "DropData":
+		*p = DropDataPermission
+	case "ReadData":
+		*p = ReadDataPermission
+	case "9", "WriteData":
+		*p = WriteDataPermission
+	case "Rebalance":
+		*p = RebalancePermission
+	case "ManageShard":
+		*p = ManageShardPermission
+	case "ManageContinuousQuery":
+		*p = ManageContinuousQueryPermission
+	case "ManageQuery":
+		*p = ManageQueryPermission
+	case "ManageSubscription":
+		*p = ManageSubscriptionPermission
+	case "Monitor":
+		*p = MonitorPermission
+	case "CopyShard":
+		*p = CopyShardPermission
+	case "KapacitorAPI":
+		*p = KapacitorAPIPermission
+	case "KapacitorConfigAPI":
+		*p = KapacitorConfigAPIPermission
+	default:
+		return fmt.Errorf("invalid permission: %s", s)
+	}
+	return nil
+}
diff --git a/services/auth/service.go b/services/auth/service.go
new file mode 100644
index 00000000..12eccf01
--- /dev/null
+++ b/services/auth/service.go
@@ -0,0 +1,867 @@
+package auth
+
+import (
+	"bytes"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"path"
+	"regexp"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/influxdata/kapacitor/auth"
+	client "github.com/influxdata/kapacitor/client/v1"
+	"github.com/influxdata/kapacitor/keyvalue"
+	"github.com/influxdata/kapacitor/services/auth/meta"
+	"github.com/influxdata/kapacitor/services/httpd"
+	"github.com/influxdata/kapacitor/services/storage"
+	"github.com/influxdata/kapacitor/tlsconfig"
+	"github.com/pkg/errors"
+
+	"golang.org/x/crypto/bcrypt"
+)
+
+const (
+	usersPath         = "/users"
+	usersPathAnchored = "/users/"
+
+	// SaltBytes is the number of bytes used for salts
+	saltBytes = 32
+
+	authCacheExpiration = time.Hour
+)
+
+type Diagnostic interface {
+	Debug(msg string, ctx ...keyvalue.T)
+}
+
+type Service struct {
+	diag   Diagnostic
+	routes []httpd.Route
+
+	StorageService interface {
+		Store(namespace string) storage.Interface
+	}
+	HTTPDService interface {
+		AddRoutes([]httpd.Route) error
+		DelRoutes([]httpd.Route)
+	}
+
+	users           UserDAO
+	userCache       UserCache
+	cacheExpiration time.Duration
+
+	bcryptCost int
+
+	// Authentication cache.
+	// Caches sha256 hashes of passwords for faster authentication
+	authCache map[string]authCred
+	authMU    sync.RWMutex
+
+	path string
+
+	retentionAutoCreate bool
+
+	// Plutonium Meta Control client
+	pmClient *meta.Client
+}
+
+type authCred struct {
+	salt    []byte
+	hash    []byte
+	expires time.Time
+}
+
+func NewService(c Config, d Diagnostic) (*Service, error) {
+	var pmClient *meta.Client
+	if c.MetaAddr != "" {
+		tlsConfig, err := tlsconfig.Create(c.MetaCA, c.MetaCert, c.MetaKey, c.MetaInsecureSkipVerify)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not create TLS config for connecting to meta service")
+		}
+		pmOpts := []meta.ClientOption{
+			meta.WithTLS(tlsConfig, c.MetaUseTLS, c.MetaInsecureSkipVerify),
+		}
+		if c.MetaUsername != "" {
+			pmOpts = append(pmOpts, meta.UseAuth(meta.BasicAuth, c.MetaUsername, c.MetaPassword, ""))
+		}
+		//TODO: when the meta client can accept an interface, pass in a logger
+		pmClient = meta.NewClient(c.MetaAddr, pmOpts...)
+	} else {
+		d.Debug("not using meta service for users, no address given")
+	}
+
+	return &Service{
+		diag:            d,
+		authCache:       make(map[string]authCred),
+		cacheExpiration: time.Duration(c.CacheExpiration),
+		bcryptCost:      c.BcryptCost,
+		pmClient:        pmClient,
+	}, nil
+}
+
+const userNamespace = "user_store"
+
+func (s *Service) Open() error {
+	if s.StorageService == nil {
+		return errors.New("missing storage service")
+	}
+	if s.HTTPDService == nil {
+		return errors.New("missing httpd service")
+	}
+	store := s.StorageService.Store(userNamespace)
+	users, err := newUserKV(store)
+	if err != nil {
+		return err
+	}
+	s.users = users
+	s.userCache = newMemUserCache(s.cacheExpiration)
+
+	// Define API routes
+	s.routes = []httpd.Route{
+		{
+			Method:      "GET",
+			Pattern:     usersPathAnchored,
+			HandlerFunc: s.handleUser,
+		},
+		{
+			Method:      "DELETE",
+			Pattern:     usersPathAnchored,
+			HandlerFunc: s.handleDeleteUser,
+		},
+		{
+			// Satisfy CORS checks.
+			Method:      "OPTIONS",
+			Pattern:     usersPathAnchored,
+			HandlerFunc: httpd.ServeOptions,
+		},
+		{
+			Method:      "PATCH",
+			Pattern:     usersPathAnchored,
+			HandlerFunc: s.handleUpdateUser,
+		},
+		{
+			Method:      "GET",
+			Pattern:     usersPath,
+			HandlerFunc: s.handleListUsers,
+		},
+		{
+			Method:      "POST",
+			Pattern:     usersPath,
+			HandlerFunc: s.handleCreateUser,
+		},
+	}
+
+	if err := s.HTTPDService.AddRoutes(s.routes); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *Service) Close() error {
+	if s.HTTPDService != nil {
+		s.HTTPDService.DelRoutes(s.routes)
+	}
+	return nil
+}
+
+func (s *Service) Authenticate(username, password string) (auth.User, error) {
+	user, err := s.User(username)
+	if err != nil {
+		return auth.User{}, err
+	}
+
+	// Check for auth cache entry first
+	s.authMU.RLock()
+	cred, ok := s.authCache[username]
+	s.authMU.RUnlock()
+
+	// this happens when influxdb Enterprise is configured to use LDAP, we need to check the password
+	// against influxdb
+	if len(user.Hash()) == 0 {
+		if err := s.pmClient.CheckPass(username, password); err != nil {
+			return auth.User{}, errors.Wrap(err, "checking password of Influxdb Enterprise meta user")
+		}
+		hash, err := bcrypt.GenerateFromPassword([]byte(password), s.bcryptCost)
+		if err != nil {
+			return auth.User{}, errors.Wrap(err, "failed to hash user password")
+		}
+		user = auth.NewUser(user.Name(), hash, user.IsAdmin(), user.Privileges())
+		s.userCache.Set(user)
+	} else if ok {
+		// verify the password using the cached salt and hash
+		if cred.expires.After(time.Now()) && bytes.Equal(s.hashWithSalt(cred.salt, password), cred.hash) {
+			return user, nil
+		}
+		// fall through to requiring a full bcrypt hash for invalid passwords
+	}
+
+	// Compare password with user hash.
+	if err := bcrypt.CompareHashAndPassword(user.Hash(), []byte(password)); err != nil {
+		s.userCache.Delete(username)
+		return auth.User{}, fmt.Errorf("failed to authenticate user")
+	}
+
+	// generate a salt and hash of the password for the cache
+	if salt, hashed, err := s.saltedHash(password); err == nil {
+		s.authMU.Lock()
+		s.authCache[username] = authCred{salt: salt, hash: hashed, expires: time.Now().Add(authCacheExpiration)}
+		s.authMU.Unlock()
+	}
+	return user, nil
+}
+
+// saltedHash returns a salt and salted hash of password
+func (s *Service) saltedHash(password string) (salt, hash []byte, err error) {
+	salt = make([]byte, saltBytes)
+	if _, err := io.ReadFull(rand.Reader, salt); err != nil {
+		return nil, nil, err
+	}
+
+	return salt, s.hashWithSalt(salt, password), nil
+}
+
+// hashWithSalt returns a salted hash of password using salt
+func (s *Service) hashWithSalt(salt []byte, password string) []byte {
+	hasher := sha256.New()
+	hasher.Write(salt)
+	hasher.Write([]byte(password))
+	return hasher.Sum(nil)
+}
+
+func (s *Service) User(username string) (auth.User, error) {
+	// Check cache first
+	cached, found := s.userCache.Get(username)
+	if found {
+		return cached, nil
+	}
+	// Find the user somewhere
+	var user User
+
+	if s.pmClient != nil {
+		// Check in InfluxDB Enterprise meta
+		pmUsers, err := s.pmClient.Users(username)
+		if err != nil {
+			return auth.User{}, errors.Wrapf(err, "retrieving user %q from InfluxDB Enterprise meta", username)
+		}
+		if len(pmUsers) == 1 {
+			user, err = s.convertFromPMUser(pmUsers[0])
+			if err != nil {
+				return auth.User{}, errors.Wrap(err, "converting from InfluxDB Enterprise meta user")
+			}
+		}
+	} else {
+		// Check in store
+		var err error
+		user, err = s.users.Get(username)
+		if err != nil {
+			return auth.User{}, errors.Wrapf(err, "retrieving user %q from store", username)
+		}
+	}
+
+	if au, err := s.convertToAuthUser(user); err != nil {
+		return auth.User{}, errors.Wrap(err, "converting from stored user")
+	} else {
+		// Populate cache with user from store
+		s.userCache.Set(au)
+		return au, nil
+	}
+}
+
+// Return a user based on the subscription token
+func (s *Service) SubscriptionUser(token string) (auth.User, error) {
+	username, err := s.subscriptionUsername(token)
+	if err != nil {
+		return auth.User{}, err
+	}
+	// Check cache first
+	cached, found := s.userCache.Get(username)
+	if found {
+		return cached, nil
+	}
+
+	// Check in store
+	user, err := s.users.Get(username)
+	if err != nil {
+		// The user doesn't exist, so the token is bad.
+		return auth.User{}, errors.New("invalid subscription token")
+	}
+
+	if au, err := s.convertToAuthUser(user); err != nil {
+		return auth.User{}, errors.Wrap(err, "converting from stored user")
+	} else {
+		// Populate cache with user from store
+		s.userCache.Set(au)
+		return au, nil
+	}
+}
+
+const (
+	// Prefix used to identfy all users created for granting subscription token access.
+	// NOTE: this is technically an invalid username which means that a user with this prefix
+	// can't be created externally.
+	subscriptionUsernamePrefix = "_sub:"
+)
+
+// Convert the token into a subscription username, if valid.
+func (s *Service) subscriptionUsername(token string) (string, error) {
+	if !validUsername.MatchString(token) {
+		return "", fmt.Errorf("token must be a valid username: %q", token)
+	}
+	return subscriptionUsernamePrefix + token, nil
+}
+
+// Convert the username into a subscription token.
+func (s *Service) subscriptionToken(username string) string {
+	return username[len(subscriptionUsernamePrefix):]
+}
+
+func (s *Service) GrantSubscriptionAccess(token, db, rp string) error {
+	username, err := s.subscriptionUsername(token)
+	if err != nil {
+		return err
+	}
+	dbResource := auth.DatabaseResource(db)
+	user := User{
+		Name:  username,
+		Admin: false,
+		Privileges: map[string][]Privilege{
+			writeResource: []Privilege{WritePrivilege},
+			pingResource:  []Privilege{AllPrivileges},
+			dbResource:    []Privilege{WritePrivilege},
+		},
+	}
+	if err := s.users.Create(user); err != nil {
+		if err == ErrUserExists {
+			if err := s.users.Replace(user); err != nil {
+				return errors.Wrap(err, "replacing existing subscription user")
+			}
+		} else {
+			return errors.Wrap(err, "creating subscription user")
+		}
+	}
+	return nil
+}
+
+func (s *Service) ListSubscriptionTokens() ([]string, error) {
+	var tokens []string
+	offset, limit := 0, 100
+	pattern := subscriptionUsernamePrefix + "*"
+	for {
+		users, err := s.users.List(pattern, offset, limit)
+		if err != nil {
+			return nil, err
+		}
+		for _, user := range users {
+			tokens = append(tokens, s.subscriptionToken(user.Name))
+		}
+
+		if len(users) != limit {
+			break
+		}
+		offset += limit
+	}
+	return tokens, nil
+}
+
+func (s *Service) RevokeSubscriptionAccess(token string) error {
+	username, err := s.subscriptionUsername(token)
+	if err != nil {
+		return err
+	}
+	s.userCache.Delete(username)
+	err = s.users.Delete(username)
+	return errors.Wrap(err, "deleteing user")
+}
+
+// Pattern for valid usernames.
+var validUsername = regexp.MustCompile(`^[-\._\p{L}0-9@]+$`)
+
+const usersBasePathAnchored = httpd.BasePath + usersPathAnchored
+
+func (s *Service) usernameFromPath(path string) (string, error) {
+	if len(path) <= len(usersBasePathAnchored) {
+		return "", errors.New("must specify username on path")
+	}
+	username := path[len(usersBasePathAnchored):]
+	return username, nil
+}
+
+func (s *Service) userLink(username string) client.Link {
+	return client.Link{Relation: client.Self, Href: path.Join(httpd.BasePath, usersPath, username)}
+}
+
+func (s *Service) handleUser(w http.ResponseWriter, r *http.Request) {
+	username, err := s.usernameFromPath(r.URL.Path)
+	if err != nil {
+		httpd.HttpError(w, err.Error(), true, http.StatusBadRequest)
+		return
+	}
+	u, err := s.users.Get(username)
+	if err != nil {
+		if err == ErrNoUserExists {
+			httpd.HttpError(w, err.Error(), true, http.StatusNotFound)
+			return
+		}
+		httpd.HttpError(w, err.Error(), true, http.StatusInternalServerError)
+		return
+	}
+	cu, err := s.convertToClientUser(u)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to convert user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(httpd.MarshalJSON(cu, true))
+}
+
+func (s *Service) handleCreateUser(w http.ResponseWriter, r *http.Request) {
+	user := client.CreateUserOptions{}
+	dec := json.NewDecoder(r.Body)
+	err := dec.Decode(&user)
+	if err != nil {
+		httpd.HttpError(w, "invalid JSON", true, http.StatusBadRequest)
+		return
+	}
+	if user.Name == "" {
+		httpd.HttpError(w, "username is required", true, http.StatusBadRequest)
+		return
+	}
+	if !validUsername.MatchString(user.Name) {
+		httpd.HttpError(w, fmt.Sprintf("username must contain only letters, numbers, '-', '.', '@' and '_'. %q", user.Name), true, http.StatusBadRequest)
+		return
+	}
+	if user.Password == "" {
+		httpd.HttpError(w, "password is required", true, http.StatusBadRequest)
+		return
+	}
+
+	// Convert Permission to Resources/Privileges
+	privileges, err := s.convertPermissions(user.Permissions)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("invalid permissions: %s", err.Error()), true, http.StatusBadRequest)
+		return
+	}
+
+	u, err := s.CreateUser(user.Name, user.Password, user.Type == client.AdminUser, privileges)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to create user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	cu, err := s.convertToClientUser(u)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to convert user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(httpd.MarshalJSON(cu, true))
+}
+
+func (s *Service) handleUpdateUser(w http.ResponseWriter, r *http.Request) {
+	username, err := s.usernameFromPath(r.URL.Path)
+	if err != nil {
+		httpd.HttpError(w, err.Error(), true, http.StatusBadRequest)
+		return
+	}
+	user := client.UpdateUserOptions{}
+	dec := json.NewDecoder(r.Body)
+	if err := dec.Decode(&user); err != nil {
+		httpd.HttpError(w, "invalid JSON", true, http.StatusBadRequest)
+		return
+	}
+
+	// If user.Permissions is nil then the client didn't list any so don't update permissions.
+	updatePrivileges := user.Permissions != nil
+
+	// Convert Permission to Resources/Privileges
+	privileges, err := s.convertPermissions(user.Permissions)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("invalid permissions: %s", err.Error()), true, http.StatusBadRequest)
+		return
+	}
+
+	u, err := s.updateUser(
+		username,
+		user.Password,
+		user.Type != client.InvalidUser,
+		user.Type == client.AdminUser,
+		updatePrivileges,
+		privileges,
+	)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to create user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	cu, err := s.convertToClientUser(u)
+	if err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to convert user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(httpd.MarshalJSON(cu, true))
+}
+
+func (s *Service) handleDeleteUser(w http.ResponseWriter, r *http.Request) {
+	username, err := s.usernameFromPath(r.URL.Path)
+	if err != nil {
+		httpd.HttpError(w, err.Error(), true, http.StatusBadRequest)
+		return
+	}
+
+	s.userCache.Delete(username)
+
+	s.authMU.Lock()
+	delete(s.authCache, username)
+	s.authMU.Unlock()
+
+	if err := s.users.Delete(username); err != nil {
+		httpd.HttpError(w, fmt.Sprintf("failed to delete user: %s", err.Error()), true, http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+var allUserFields = []string{
+	"link",
+	"name",
+	"type",
+	"permissions",
+}
+
+func (s *Service) handleListUsers(w http.ResponseWriter, r *http.Request) {
+	pattern := r.URL.Query().Get("pattern")
+	fields := r.URL.Query()["fields"]
+	if len(fields) == 0 {
+		fields = allUserFields
+	} else {
+		// Always return ID field
+		fields = append(fields, "name", "link")
+	}
+
+	var err error
+	offset := int64(0)
+	offsetStr := r.URL.Query().Get("offset")
+	if offsetStr != "" {
+		offset, err = strconv.ParseInt(offsetStr, 10, 64)
+		if err != nil {
+			httpd.HttpError(w, fmt.Sprintf("invalid offset parameter %q must be an integer: %s", offsetStr, err), true, http.StatusBadRequest)
+		}
+	}
+
+	limit := int64(100)
+	limitStr := r.URL.Query().Get("limit")
+	if limitStr != "" {
+		limit, err = strconv.ParseInt(limitStr, 10, 64)
+		if err != nil {
+			httpd.HttpError(w, fmt.Sprintf("invalid limit parameter %q must be an integer: %s", limitStr, err), true, http.StatusBadRequest)
+		}
+	}
+
+	rawUsers, err := s.users.List(pattern, int(offset), int(limit))
+	users := make([]map[string]interface{}, len(rawUsers))
+
+	for i, user := range rawUsers {
+		users[i] = make(map[string]interface{}, len(fields))
+		for _, field := range fields {
+			var value interface{}
+			switch field {
+			case "name":
+				value = user.Name
+			case "link":
+				value = s.userLink(user.Name)
+			case "type":
+				value = client.NormalUser
+				if user.Admin {
+					value = client.AdminUser
+				}
+			case "permissions":
+				value = s.convertPrivleges(user.Privileges)
+			default:
+				httpd.HttpError(w, fmt.Sprintf("unsupported field %q", field), true, http.StatusBadRequest)
+				return
+			}
+			users[i][field] = value
+		}
+	}
+
+	type response struct {
+		Users []map[string]interface{} `json:"users"`
+	}
+
+	w.Write(httpd.MarshalJSON(response{users}, true))
+}
+
+var (
+	rootResource   = auth.APIResource("/")
+	writeResource  = auth.APIResource("/write")
+	pingResource   = auth.APIResource("/ping")
+	configResource = auth.APIResource("/config")
+)
+
+// Map permissions to a set of resources and privileges
+func (s *Service) convertPermissions(perms []client.Permission) (map[string][]Privilege, error) {
+	privileges := make(map[string][]Privilege, len(perms))
+	for _, perm := range perms {
+		switch perm {
+		case client.NoPermissions:
+		case client.APIPermission:
+			privileges[rootResource] = []Privilege{AllPrivileges}
+			// Subtractive permission, only add it if something else doesn't already exist.
+			if _, ok := privileges[writeResource]; !ok {
+				privileges[writeResource] = []Privilege{NoPrivileges}
+			}
+			// Do not give config API access unless specificaly granted
+			if _, ok := privileges[configResource]; !ok {
+				privileges[configResource] = []Privilege{NoPrivileges}
+			}
+		case client.ConfigAPIPermission:
+			privileges[pingResource] = []Privilege{AllPrivileges}
+			privileges[configResource] = []Privilege{AllPrivileges}
+		case client.WritePointsPermission:
+			privileges[pingResource] = []Privilege{AllPrivileges}
+			privileges[writeResource] = []Privilege{AllPrivileges}
+		case client.AllPermissions:
+			privileges[rootResource] = []Privilege{AllPrivileges}
+			privileges[configResource] = []Privilege{AllPrivileges}
+			privileges[writeResource] = []Privilege{AllPrivileges}
+		default:
+			return nil, fmt.Errorf("unknown permission %v", perm)
+		}
+	}
+	return privileges, nil
+}
+
+// Map a set of resources and privileges to permissions
+func (s *Service) convertPrivleges(privileges map[string][]Privilege) []client.Permission {
+	hasAPI := false
+	if ps, ok := privileges[rootResource]; ok {
+		for _, p := range ps {
+			if p == AllPrivileges {
+				hasAPI = true
+			}
+		}
+	}
+	hasConfig := false
+	if ps, ok := privileges[configResource]; ok {
+		for _, p := range ps {
+			if p == AllPrivileges {
+				hasConfig = true
+			}
+		}
+	}
+	hasWrite := false
+	if ps, ok := privileges[writeResource]; ok {
+		for _, p := range ps {
+			if p == AllPrivileges {
+				hasWrite = true
+			}
+		}
+	}
+	if hasAPI && hasWrite && hasConfig {
+		return []client.Permission{client.AllPermissions}
+	}
+	perms := make([]client.Permission, 0)
+	if hasAPI {
+		perms = append(perms, client.APIPermission)
+	}
+	if hasConfig {
+		perms = append(perms, client.ConfigAPIPermission)
+	}
+	if hasWrite {
+		perms = append(perms, client.WritePointsPermission)
+	}
+	return perms
+}
+
+// Map a set of InfluxDB Enterprise meta permissions to a set of privileges.
+func convertPMPermissions(scopedPerms map[string][]meta.Permission) map[string][]Privilege {
+	privileges := make(map[string][]Privilege, len(scopedPerms))
+	for scope, perms := range scopedPerms {
+		dbResource := auth.DatabaseResource(scope)
+		for _, perm := range perms {
+			switch perm {
+			case meta.KapacitorAPIPermission:
+				privileges[rootResource] = []Privilege{AllPrivileges}
+				// Do not give config API access unless specifically granted
+				if _, ok := privileges[configResource]; !ok {
+					privileges[configResource] = []Privilege{NoPrivileges}
+				}
+			case meta.KapacitorConfigAPIPermission:
+				privileges[pingResource] = []Privilege{AllPrivileges}
+				privileges[configResource] = []Privilege{AllPrivileges}
+			case meta.WriteDataPermission:
+				privileges[dbResource] = append(privileges[dbResource], WritePrivilege)
+				privileges[writeResource] = []Privilege{WritePrivilege}
+				privileges[pingResource] = []Privilege{AllPrivileges}
+			case meta.ReadDataPermission:
+				privileges[pingResource] = []Privilege{AllPrivileges}
+				privileges[dbResource] = append(privileges[dbResource], ReadPrivilege)
+			default:
+				// Ignore, user can have permissions for InfluxDB related actions.
+			}
+		}
+	}
+	return privileges
+}
+
+func (s *Service) CreateUser(username, password string, admin bool, privileges map[string][]Privilege) (User, error) {
+	// Check if user exists
+	_, err := s.users.Get(username)
+	if err != ErrNoUserExists {
+		return User{}, errors.Wrap(err, "creating user")
+	}
+
+	// Hash the password before serializing it.
+	hash, err := bcrypt.GenerateFromPassword([]byte(password), s.bcryptCost)
+	if err != nil {
+		return User{}, errors.Wrap(err, "hashing password")
+	}
+	u := User{
+		Name:       username,
+		Hash:       hash,
+		Admin:      admin,
+		Privileges: privileges,
+	}
+
+	if err := s.users.Create(u); err != nil {
+		return User{}, errors.Wrap(err, "saving user")
+	}
+	// Populate user cache
+	if au, err := s.convertToAuthUser(u); err == nil {
+		s.userCache.Set(au)
+	}
+	return u, nil
+}
+
+func (s *Service) updateUser(username, password string, updateAdmin, admin, updatePrivileges bool, privileges map[string][]Privilege) (User, error) {
+	// Check if user exists
+	u, err := s.users.Get(username)
+	if err != nil {
+		return User{}, errors.Wrap(err, "updating user")
+	}
+
+	if password != "" {
+		// Hash the password before serializing it.
+		hash, err := bcrypt.GenerateFromPassword([]byte(password), s.bcryptCost)
+		if err != nil {
+			return User{}, errors.Wrap(err, "hashing password")
+		}
+		u.Hash = hash
+	}
+
+	if updatePrivileges {
+		u.Privileges = privileges
+	}
+	if updateAdmin {
+		u.Admin = admin
+		if admin {
+			// Zero user privileges since now its an admin.
+			u.Privileges = nil
+		}
+	}
+
+	if err := s.users.Replace(u); err != nil {
+		return User{}, errors.Wrap(err, "saving user")
+	}
+	// Populate user cache
+	if au, err := s.convertToAuthUser(u); err == nil {
+		s.userCache.Set(au)
+	}
+	return u, nil
+}
+
+func (s *Service) DeleteUser(username string) {
+	s.users.Delete(username)
+	s.userCache.Delete(username)
+}
+
+// Convert a user from the store into an auth.User.
+func (s *Service) convertToAuthUser(u User) (auth.User, error) {
+	privileges := make(map[string][]auth.Privilege, len(u.Privileges))
+	for r, ps := range u.Privileges {
+		for _, p := range ps {
+			var priv auth.Privilege
+			switch p {
+			case NoPrivileges:
+				priv = auth.NoPrivileges
+			case ReadPrivilege:
+				priv = auth.ReadPrivilege
+			case WritePrivilege:
+				priv = auth.WritePrivilege
+			case DeletePrivilege:
+				priv = auth.DeletePrivilege
+			case AllPrivileges:
+				priv = auth.AllPrivileges
+			default:
+				return auth.User{}, fmt.Errorf("unknown Privilege %v", p)
+			}
+			privileges[r] = append(privileges[r], priv)
+		}
+	}
+	au := auth.NewUser(u.Name, u.Hash, u.Admin, privileges)
+	return au, nil
+}
+
+// Convert an auth.User into a user for the store.
+func (s *Service) convertFromAuthUser(au auth.User) (User, error) {
+	u := User{
+		Name:  au.Name(),
+		Admin: au.IsAdmin(),
+		Hash:  au.Hash(),
+	}
+	if !u.Admin {
+		privileges := au.Privileges()
+		u.Privileges = make(map[string][]Privilege, len(privileges))
+		for r, ps := range privileges {
+			for _, p := range ps {
+				var priv Privilege
+				switch p {
+				case auth.NoPrivileges:
+					priv = NoPrivileges
+				case auth.ReadPrivilege:
+					priv = ReadPrivilege
+				case auth.WritePrivilege:
+					priv = WritePrivilege
+				case auth.DeletePrivilege:
+					priv = DeletePrivilege
+				case auth.AllPrivileges:
+					priv = AllPrivileges
+				default:
+					return User{}, fmt.Errorf("unknown auth.Privilege %v", p)
+				}
+				u.Privileges[r] = append(u.Privileges[r], priv)
+			}
+		}
+	}
+	return u, nil
+}
+
+// Convert a meta.User into a user for the store.
+func (s *Service) convertFromPMUser(mu meta.User) (User, error) {
+	u := User{
+		Name:       mu.Name,
+		Hash:       []byte(mu.Hash),
+		Privileges: convertPMPermissions(mu.Permissions),
+	}
+	return u, nil
+}
+
+// Convert a user from the store into an client.User.
+func (s *Service) convertToClientUser(u User) (client.User, error) {
+	ut := client.NormalUser
+	if u.Admin {
+		ut = client.AdminUser
+	}
+	perms := s.convertPrivleges(u.Privileges)
+	return client.User{
+		Link:        s.userLink(u.Name),
+		Name:        u.Name,
+		Type:        ut,
+		Permissions: perms,
+	}, nil
+}
diff --git a/services/bigpanda/config.go b/services/bigpanda/config.go
index 292a2787..94e2fe30 100644
--- a/services/bigpanda/config.go
+++ b/services/bigpanda/config.go
@@ -1,8 +1,9 @@
 package bigpanda
 
 import (
-	"github.com/pkg/errors"
 	"net/url"
+
+	"github.com/pkg/errors"
 )
 
 const (
diff --git a/services/bigpanda/service.go b/services/bigpanda/service.go
index 805c0543..d7b2aae4 100644
--- a/services/bigpanda/service.go
+++ b/services/bigpanda/service.go
@@ -5,8 +5,6 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
-	khttp "github.com/influxdata/kapacitor/http"
-	"github.com/influxdata/kapacitor/models"
 	"html"
 	"io/ioutil"
 	"net/http"
@@ -16,7 +14,9 @@ import (
 	"time"
 
 	"github.com/influxdata/kapacitor/alert"
+	khttp "github.com/influxdata/kapacitor/http"
 	"github.com/influxdata/kapacitor/keyvalue"
+	"github.com/influxdata/kapacitor/models"
 	"github.com/pkg/errors"
 )
 
diff --git a/services/diagnostic/handlers.go b/services/diagnostic/handlers.go
index f9209dd3..37c53bc4 100644
--- a/services/diagnostic/handlers.go
+++ b/services/diagnostic/handlers.go
@@ -43,6 +43,7 @@ import (
 	"github.com/influxdata/kapacitor/services/telegram"
 	"github.com/influxdata/kapacitor/services/udp"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/influxdata/kapacitor/udf"
 	"github.com/influxdata/kapacitor/uuid"
 	plog "github.com/prometheus/common/log"
@@ -1033,6 +1034,17 @@ func (h *NoAuthHandler) FakedSubscriptionUserToken() {
 	h.l.Info("using noauth auth backend. Faked authentication for subscription user token")
 }
 
+type AuthHandler struct {
+	l Logger
+}
+
+func (h *AuthHandler) Error(msg string, err error, ctx ...keyvalue.T) {
+	Err(h.l, msg, err, ctx)
+}
+func (h *AuthHandler) Debug(msg string, ctx ...keyvalue.T) {
+	Debug(h.l, msg, ctx)
+}
+
 // Stats handler
 
 type StatsHandler struct {
@@ -1401,3 +1413,20 @@ func (h *ServiceNowHandler) WithContext(ctx ...keyvalue.T) servicenow.Diagnostic
 func (h *ServiceNowHandler) Error(msg string, err error) {
 	h.l.Error(msg, Error(err))
 }
+
+// Zenoss handler
+type ZenossHandler struct {
+	l Logger
+}
+
+func (h *ZenossHandler) WithContext(ctx ...keyvalue.T) zenoss.Diagnostic {
+	fields := logFieldsFromContext(ctx)
+
+	return &ZenossHandler{
+		l: h.l.With(fields...),
+	}
+}
+
+func (h *ZenossHandler) Error(msg string, err error) {
+	h.l.Error(msg, Error(err))
+}
diff --git a/services/diagnostic/service.go b/services/diagnostic/service.go
index e009fc33..b92c2528 100644
--- a/services/diagnostic/service.go
+++ b/services/diagnostic/service.go
@@ -342,6 +342,12 @@ func (s *Service) NewNoAuthHandler() *NoAuthHandler {
 	}
 }
 
+func (s *Service) NewAuthHandler() *AuthHandler {
+	return &AuthHandler{
+		l: s.Logger.With(String("service", "auth")),
+	}
+}
+
 func (s *Service) NewStatsHandler() *StatsHandler {
 	return &StatsHandler{
 		l: s.Logger.With(String("service", "stats")),
@@ -490,3 +496,9 @@ func (s *Service) NewServiceNowHandler() *ServiceNowHandler {
 		l: s.Logger.With(String("service", "serviceNow")),
 	}
 }
+
+func (s *Service) NewZenossHandler() *ZenossHandler {
+	return &ZenossHandler{
+		l: s.Logger.With(String("service", "zenoss")),
+	}
+}
diff --git a/services/kafka/config.go b/services/kafka/config.go
index 5dad033c..804b5ce7 100644
--- a/services/kafka/config.go
+++ b/services/kafka/config.go
@@ -76,7 +76,32 @@ func (c *Config) ApplyConditionalDefaults() {
 	}
 }
 
-func (c Config) WriterConfig(diagnostic Diagnostic) (kafka.WriterConfig, error) {
+type WriteTarget struct {
+	Topic              string
+	PartitionById      bool
+	PartitionAlgorithm string
+}
+
+func (c Config) WriterConfig(diagnostic Diagnostic, target WriteTarget) (kafka.WriterConfig, error) {
+	if target.Topic == "" {
+		return kafka.WriterConfig{}, errors.New("topic must not be empty")
+	}
+	var balancer kafka.Balancer
+	if target.PartitionById {
+		switch target.PartitionAlgorithm {
+		case "crc32c", "":
+			balancer = &kafka.CRC32Balancer{}
+		case "murmur2":
+			balancer = &kafka.Murmur2Balancer{}
+		case "fnv-1a":
+			balancer = &kafka.Hash{}
+		default:
+			return kafka.WriterConfig{}, fmt.Errorf("invalid partition algorithm: %q", target.PartitionAlgorithm)
+		}
+	} else {
+		balancer = &kafka.LeastBytes{}
+	}
+
 	var tlsCfg *tls.Config
 	if c.UseSSL {
 		t, err := tlsconfig.Create(c.SSLCA, c.SSLCert, c.SSLKey, c.InsecureSkipVerify)
@@ -90,9 +115,8 @@ func (c Config) WriterConfig(diagnostic Diagnostic) (kafka.WriterConfig, error)
 		TLS:     tlsCfg,
 	}
 
-	return kafka.WriterConfig{
+	baseConfig := kafka.WriterConfig{
 		Brokers:      c.Brokers,
-		Balancer:     &kafka.LeastBytes{},
 		Dialer:       dialer,
 		ReadTimeout:  time.Duration(c.Timeout),
 		WriteTimeout: time.Duration(c.Timeout),
@@ -105,7 +129,11 @@ func (c Config) WriterConfig(diagnostic Diagnostic) (kafka.WriterConfig, error)
 		ErrorLogger: kafka.LoggerFunc(func(s string, x ...interface{}) {
 			diagnostic.Error("kafka client error", fmt.Errorf(s, x...))
 		}),
-	}, nil
+		Topic:    target.Topic,
+		Balancer: balancer,
+	}
+
+	return baseConfig, nil
 }
 
 type Configs []Config
diff --git a/services/kafka/service.go b/services/kafka/service.go
index 928d73f1..f588a90d 100644
--- a/services/kafka/service.go
+++ b/services/kafka/service.go
@@ -138,8 +138,8 @@ func NewCluster(c Config) *Cluster {
 	}
 }
 
-func (c *Cluster) WriteMessage(diagnostic Diagnostic, topic string, key, msg []byte) error {
-	w, err := c.writer(topic, diagnostic)
+func (c *Cluster) WriteMessage(diagnostic Diagnostic, target WriteTarget, key, msg []byte) error {
+	w, err := c.writer(target, diagnostic)
 	if err != nil {
 		return err
 	}
@@ -149,7 +149,8 @@ func (c *Cluster) WriteMessage(diagnostic Diagnostic, topic string, key, msg []b
 	})
 }
 
-func (c *Cluster) writer(topic string, diagnostic Diagnostic) (*writer, error) {
+func (c *Cluster) writer(target WriteTarget, diagnostic Diagnostic) (*writer, error) {
+	topic := target.Topic
 	c.mu.RLock()
 	w, ok := c.writers[topic]
 	c.mu.RUnlock()
@@ -158,14 +159,10 @@ func (c *Cluster) writer(topic string, diagnostic Diagnostic) (*writer, error) {
 		defer c.mu.Unlock()
 		w, ok = c.writers[topic]
 		if !ok {
-			wc, err := c.cfg.WriterConfig(diagnostic)
+			wc, err := c.cfg.WriterConfig(diagnostic, target)
 			if err != nil {
 				return nil, err
 			}
-			if topic == "" {
-				return nil, errors.New("topic must not be empty")
-			}
-			wc.Topic = topic
 			kw := kafka.NewWriter(wc)
 			// Create new writer
 			w = &writer{
@@ -329,21 +326,23 @@ func (s *Service) Test(options interface{}) error {
 	if !ok {
 		return fmt.Errorf("unknown cluster %q", o.Cluster)
 	}
-	return c.WriteMessage(s.diag, o.Topic, []byte(o.Key), []byte(o.Message))
+	return c.WriteMessage(s.diag, WriteTarget{Topic: o.Topic}, []byte(o.Key), []byte(o.Message))
 }
 
 type HandlerConfig struct {
-	Cluster  string `mapstructure:"cluster"`
-	Topic    string `mapstructure:"topic"`
-	Template string `mapstructure:"template"`
+	Cluster              string `mapstructure:"cluster"`
+	Topic                string `mapstructure:"topic"`
+	Template             string `mapstructure:"template"`
+	DisablePartitionById bool   `mapstructure:"disablePartitionById"`
+	PartitionAlgorithm   string `mapstructure:"partitionAlgorithm"`
 }
 
 type handler struct {
 	s *Service
 
-	cluster  *Cluster
-	topic    string
-	template *template.Template
+	cluster     *Cluster
+	writeTarget WriteTarget
+	template    *template.Template
 
 	diag Diagnostic
 }
@@ -365,9 +364,13 @@ func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) (alert.Handler, er
 	diag := s.diag.WithContext(ctx...)
 
 	return &handler{
-		s:        s,
-		cluster:  cluster,
-		topic:    c.Topic,
+		s:       s,
+		cluster: cluster,
+		writeTarget: WriteTarget{
+			Topic:              c.Topic,
+			PartitionById:      !c.DisablePartitionById,
+			PartitionAlgorithm: c.PartitionAlgorithm,
+		},
 		template: t,
 		diag:     diag,
 	}, nil
@@ -378,7 +381,7 @@ func (h *handler) Handle(event alert.Event) {
 	if err != nil {
 		h.diag.Error("failed to prepare kafka message body", err)
 	}
-	if err := h.cluster.WriteMessage(h.diag, h.topic, []byte(event.State.ID), body); err != nil {
+	if err := h.cluster.WriteMessage(h.diag, h.writeTarget, []byte(event.State.ID), body); err != nil {
 		h.diag.Error("failed to write message to kafka", err)
 	}
 }
diff --git a/services/storage/version_easyjson.go b/services/storage/version_easyjson.go
index bcb42c45..ecfbc761 100644
--- a/services/storage/version_easyjson.go
+++ b/services/storage/version_easyjson.go
@@ -4,6 +4,7 @@ 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/zenoss/config.go b/services/zenoss/config.go
new file mode 100644
index 00000000..5b87b747
--- /dev/null
+++ b/services/zenoss/config.go
@@ -0,0 +1,84 @@
+package zenoss
+
+import (
+	"net/url"
+
+	"github.com/influxdata/kapacitor/alert"
+	"github.com/pkg/errors"
+)
+
+type SeverityMap struct {
+	// OK level mapping to severity (number or string)
+	OK interface{} `toml:"OK" json:"ok"`
+	// Info level mapping to severity (number or string)
+	Info interface{} `toml:"Info" json:"info"`
+	// Warning level mapping to severity (number or string)
+	Warning interface{} `toml:"Warning" json:"warning"`
+	// Critical level mapping to severity (number or string)
+	Critical interface{} `toml:"Critical" json:"critical"`
+}
+
+func (s *SeverityMap) ValueFor(level alert.Level) interface{} {
+	var severity interface{}
+	switch level {
+	case alert.OK:
+		severity = s.OK
+	case alert.Info:
+		severity = s.Info
+	case alert.Warning:
+		severity = s.Warning
+	case alert.Critical:
+		severity = s.Critical
+	}
+	return severity
+}
+
+type Config struct {
+	// Whether Zenoss integration is enabled.
+	Enabled bool `toml:"enabled" override:"enabled"`
+	// Zenoss events API URL.
+	URL string `toml:"url" override:"url"`
+	// ServiceNow authentication username.
+	Username string `toml:"username" override:"username"`
+	// ServiceNow authentication password.
+	Password string `toml:"password" override:"password,redact"`
+	// Action (router name).
+	Action string `toml:"action" override:"action"`
+	// Router method.
+	Method string `toml:"method" override:"method"`
+	// Event type.
+	Type string `toml:"type" override:"type"`
+	// Event TID.
+	TID int64 `toml:"tid" override:"tid"`
+	// Level to severity map.
+	SeverityMap SeverityMap `toml:"severity-map" override:"severity-map"`
+	// Whether all alerts should automatically post to ServiceNow.
+	Global bool `toml:"global" override:"global"`
+	// Whether all alerts should automatically use stateChangesOnly mode.
+	// Only applies if global is also set.
+	StateChangesOnly bool `toml:"state-changes-only" override:"state-changes-only"`
+}
+
+func NewConfig() Config {
+	return Config{
+		URL:         "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router",
+		Action:      "EventsRouter",
+		Method:      "add_event",
+		Type:        "rpc",
+		TID:         1,
+		SeverityMap: SeverityMap{OK: "Clear", Info: "Info", Warning: "Warning", Critical: "Critical"},
+	}
+}
+
+func (c Config) Validate() error {
+	if c.Enabled && c.URL == "" {
+		return errors.New("must specify events URL")
+	}
+	if _, err := url.Parse(c.URL); err != nil {
+		return errors.Wrapf(err, "invalid url %q", c.URL)
+	}
+	if c.SeverityMap.OK == nil || c.SeverityMap.Info == nil || c.SeverityMap.Warning == nil || c.SeverityMap.Critical == nil {
+		return errors.New("invalid severity mapping")
+	}
+	return nil
+}
diff --git a/services/zenoss/service.go b/services/zenoss/service.go
new file mode 100644
index 00000000..c5087e21
--- /dev/null
+++ b/services/zenoss/service.go
@@ -0,0 +1,475 @@
+package zenoss
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math"
+	"net/http"
+	neturl "net/url"
+	"strings"
+	"sync/atomic"
+	text "text/template"
+	"time"
+
+	"github.com/influxdata/kapacitor/alert"
+	"github.com/influxdata/kapacitor/keyvalue"
+	"github.com/pkg/errors"
+)
+
+//const usualCutoff = 100
+
+type Diagnostic interface {
+	WithContext(ctx ...keyvalue.T) Diagnostic
+	Error(msg string, err error)
+}
+
+type Service struct {
+	configValue atomic.Value
+	diag        Diagnostic
+}
+
+func NewService(c Config, d Diagnostic) *Service {
+	s := &Service{
+		diag: d,
+	}
+	s.configValue.Store(c)
+
+	return s
+}
+
+func (s *Service) Open() error {
+	return nil
+}
+
+func (s *Service) Close() error {
+	return nil
+}
+
+func (s *Service) config() Config {
+	return s.configValue.Load().(Config)
+}
+
+func (s *Service) Update(newConfig []interface{}) error {
+	if l := len(newConfig); l != 1 {
+		return fmt.Errorf("expected only one new config object, got %d", l)
+	}
+	if c, ok := newConfig[0].(Config); !ok {
+		return fmt.Errorf("expected config object to be of type %T, got %T", c, newConfig[0])
+	} else {
+		s.configValue.Store(c)
+	}
+
+	return nil
+}
+
+func (s *Service) Global() bool {
+	return s.config().Global
+}
+
+func (s *Service) StateChangesOnly() bool {
+	return s.config().StateChangesOnly
+}
+
+type testOptions struct {
+	AlertID       string                 `json:"alert_id"`
+	Level         alert.Level            `json:"level"`
+	Message       string                 `json:"message"`
+	Action        string                 `json:"action"`
+	Method        string                 `json:"method"`
+	Type          string                 `json:"type"`
+	TID           int64                  `json:"tid"`
+	Summary       string                 `json:"summary"`
+	Device        string                 `json:"device"`
+	Component     string                 `json:"component"`
+	EventClassKey string                 `json:"event_class_key"`
+	EventClass    string                 `json:"event_class"`
+	Collector     string                 `json:"collector"`
+	CustomFields  map[string]interface{} `json:"custom_fields"`
+}
+
+func (s *Service) TestOptions() interface{} {
+	return &testOptions{
+		AlertID:      "1001",
+		Level:        alert.Critical,
+		Message:      "test zenoss message",
+		Action:       "EventsRouter",
+		Method:       "add_event",
+		Type:         "rpc",
+		TID:          1,
+		CustomFields: map[string]interface{}{},
+	}
+}
+
+func (s *Service) Test(options interface{}) error {
+	o, ok := options.(*testOptions)
+	if !ok {
+		return fmt.Errorf("unexpected options type %T", options)
+	}
+	c := s.config()
+	hc := &HandlerConfig{
+		Action:        o.Action,
+		Method:        o.Method,
+		Type:          o.Type,
+		TID:           o.TID,
+		Summary:       o.Summary,
+		Device:        o.Device,
+		Component:     o.Component,
+		EventClassKey: o.EventClassKey,
+		EventClass:    o.EventClass,
+		Collector:     o.Collector,
+		CustomFields:  o.CustomFields,
+	}
+	data := &alert.EventData{
+		Fields: map[string]interface{}{},
+		Tags:   map[string]string{},
+	}
+	state := &alert.EventState{
+		ID:      o.AlertID,
+		Level:   o.Level,
+		Message: o.Message,
+	}
+
+	return s.Alert(c.URL, state, data, hc)
+}
+
+func (s *Service) Alert(url string, state *alert.EventState, data *alert.EventData, hc *HandlerConfig) error {
+	postUrl, postBody, err := s.preparePost(url, state, data, hc)
+	if err != nil {
+		return err
+	}
+
+	req, err := http.NewRequest("POST", postUrl, postBody)
+	if err != nil {
+		return err
+	}
+
+	req.Header.Add("Content-Type", "application/json")
+	c := s.config()
+	if c.Username != "" && c.Password != "" {
+		req.SetBasicAuth(c.Username, c.Password)
+	}
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+
+		type response struct {
+			Error string `json:"error"`
+		}
+		r := &response{Error: fmt.Sprintf("failed to understand Zenoss response. code: %d content: %s", resp.StatusCode, string(body))}
+		b := bytes.NewReader(body)
+		dec := json.NewDecoder(b)
+		dec.Decode(r)
+
+		return errors.New(r.Error)
+	}
+
+	return nil
+}
+
+// EventData represents Zenoss event data. All fields should be present in JSON.
+type EventData struct {
+	Summary    string      `json:"summary"`
+	Device     string      `json:"device"`
+	Component  string      `json:"component"`
+	Severity   interface{} `json:"severity"`
+	EvClassKey string      `json:"evclasskey"`
+	EvClass    string      `json:"evclass"`
+	Collector  string      `json:"collector,omitempty"`
+	Message    string      `json:"message,omitempty"`
+}
+
+// Event is a structure representing Zenoss event.
+type Event struct {
+	Action string                   `json:"action"`
+	Method string                   `json:"method"`
+	Data   []map[string]interface{} `json:"data"`
+	Type   string                   `json:"type"`
+	TID    int64                    `json:"tid"`
+}
+
+func (s *Service) preparePost(url string, state *alert.EventState, data *alert.EventData, hc *HandlerConfig) (string, io.Reader, error) {
+	c := s.config()
+	if !c.Enabled {
+		return "", nil, errors.New("service is not enabled")
+	}
+
+	if url == "" {
+		url = c.URL
+	}
+	u, err := neturl.Parse(url)
+	if err != nil {
+		return "", nil, err
+	}
+
+	// fallback to config values for handler options not set
+	action := hc.Action
+	if action == "" {
+		action = c.Action
+	}
+	method := hc.Method
+	if method == "" {
+		method = c.Method
+	}
+	typ := hc.Type
+	if typ == "" {
+		typ = c.Type
+	}
+	tid := hc.TID
+	if tid == 0 {
+		tid = c.TID
+	}
+
+	cutoff := func(text string, max int) string {
+		return text[:int(math.Min(float64(max), float64(len(text))))]
+	}
+
+	// resolve templates for event standard data fields
+	var buffer bytes.Buffer
+	info := struct {
+		Name        string
+		TaskName    string
+		ID          string
+		Message     string
+		Details     string
+		Time        time.Time
+		Duration    time.Duration
+		Level       alert.Level
+		Recoverable bool
+		Tags        map[string]string
+	}{
+		data.Name,
+		data.TaskName,
+		state.ID,
+		state.Message,
+		state.Details,
+		state.Time,
+		state.Duration,
+		state.Level,
+		data.Recoverable,
+		data.Tags,
+	}
+	render := func(name, template string) (string, error) {
+		if template != "" {
+			buffer.Reset()
+			templateImpl, err := text.New(name).Parse(template)
+			if err != nil {
+				return "", err
+			}
+			templateImpl.Execute(&buffer, &info)
+			if err != nil {
+				return "", err
+			}
+			return buffer.String(), nil
+		}
+		return "", nil
+	}
+	summary, err := render("summary", hc.Summary)
+	if err != nil {
+		return "", nil, err
+	}
+	device, err := render("device", hc.Device)
+	if err != nil {
+		return "", nil, err
+	}
+	component, err := render("component", hc.Component)
+	if err != nil {
+		return "", nil, err
+	}
+	evclasskey, err := render("evclasskey", hc.EventClassKey)
+	if err != nil {
+		return "", nil, err
+	}
+	evclass, err := render("evclass", hc.EventClass)
+	if err != nil {
+		return "", nil, err
+	}
+	message, err := render("message", hc.Message)
+	if err != nil {
+		return "", nil, err
+	}
+	collector, err := render("collector", hc.Collector)
+	if err != nil {
+		return "", nil, err
+	}
+	severity := c.SeverityMap.ValueFor(state.Level)
+
+	if summary == "" { // fallback to default (ie. alert message)
+		summary = state.Message
+	}
+
+	eventData := &EventData{
+		Summary:    cutoff(summary, 256),
+		Device:     device,
+		Component:  component,
+		Severity:   severity,
+		EvClassKey: evclasskey,
+		EvClass:    evclass,
+		Message:    cutoff(message, 4096),
+		Collector:  collector,
+	}
+	eventDataMap, err := s.toMap(eventData)
+	if err != nil {
+		return "", nil, err
+	}
+
+	event := &Event{
+		Action: action,
+		Method: method,
+		Data: []map[string]interface{}{
+			eventDataMap,
+		},
+		Type: typ,
+		TID:  tid,
+	}
+
+	if len(hc.CustomFields) > 0 {
+		customDataMap := make(map[string]interface{})
+		for key, value := range hc.CustomFields {
+			switch v := value.(type) {
+			case string:
+				// resolve templates
+				rv, err := render("customData."+key, v)
+				if err != nil {
+					return "", nil, err
+				}
+				customDataMap[key] = rv
+			default:
+				customDataMap[key] = v
+			}
+		}
+		s.addCustomData(eventDataMap, customDataMap)
+	}
+
+	postBytes, err := json.Marshal(event)
+	if err != nil {
+		return "", nil, errors.Wrap(err, "error marshaling event struct")
+	}
+
+	return u.String(), bytes.NewBuffer(postBytes), nil
+}
+
+func (s *Service) addCustomData(basicData, customData map[string]interface{}) {
+	// add custom fields
+	for key, value := range customData {
+		var cValue interface{}
+		switch v := value.(type) {
+		case string:
+			if strings.HasPrefix(v, "{") || strings.HasPrefix(v, "[") { // can be JSON
+				err := json.Unmarshal([]byte(v), &cValue)
+				if err != nil { // guess not, it is just a text
+					cValue = v
+				}
+			} else {
+				cValue = v
+			}
+		default:
+			cValue = v
+		}
+		basicData[key] = cValue
+	}
+}
+
+func (s *Service) toMap(data *EventData) (map[string]interface{}, error) {
+	dataBytes, err := json.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	m := make(map[string]interface{})
+	err = json.Unmarshal(dataBytes, &m)
+	if err != nil {
+		return nil, err
+	}
+
+	return m, nil
+}
+
+type HandlerConfig struct {
+	// web service URL used to post messages.
+	// If empty uses the service URL from the configuration.
+	URL string `mapstructure:"url"`
+
+	// Username for BASIC authentication.
+	// If empty uses username from the configuration.
+	Username string `mapstructure:"username"`
+
+	// Password for BASIC authentication.
+	// If empty uses password from the configuration.
+	Password string `mapstructure:"password"`
+
+	// Action (router name).
+	// If empty uses action from the configuration.
+	Action string `mapstructure:"action"`
+
+	// Router method.
+	// If empty uses method from the configuration.
+	Method string `mapstructure:"method"`
+
+	// Event type.
+	// If empty uses type from the configuration.
+	Type string `mapstructure:"type"`
+
+	// Event TID.
+	// If empty uses TID from the configuration.
+	TID int64 `mapstructure:"tid"`
+
+	// Summary of the event.
+	Summary string `json:"summary"`
+
+	// Device related to the event.
+	Device string `json:"device"`
+
+	// Component related to the event.
+	Component string `json:"component"`
+
+	// Event Class Key of the event.
+	EventClassKey string `json:"evclasskey"`
+
+	// Event Class of the event.
+	EventClass string `json:"evclass"`
+
+	// Collector (typically IP or hostname).
+	Collector string `json:"collector"`
+
+	// Message related to the event.
+	Message string `json:"message"`
+
+	// Custom fields is a map of key value pairs.
+	CustomFields map[string]interface{} `mapstructure:"customField"`
+}
+
+type handler struct {
+	s    *Service
+	c    HandlerConfig
+	diag Diagnostic
+}
+
+func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) alert.Handler {
+	return &handler{
+		s:    s,
+		c:    c,
+		diag: s.diag.WithContext(ctx...),
+	}
+}
+
+func (h *handler) Handle(event alert.Event) {
+	if err := h.s.Alert(
+		h.c.URL,
+		&event.State,
+		&event.Data,
+		&h.c,
+	); err != nil {
+		h.diag.Error("failed to send event to Zenoss", err)
+	}
+}
diff --git a/task_master.go b/task_master.go
index 4ba9108e..694bda9f 100644
--- a/task_master.go
+++ b/task_master.go
@@ -43,6 +43,7 @@ import (
 	"github.com/influxdata/kapacitor/services/teams"
 	"github.com/influxdata/kapacitor/services/telegram"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/services/zenoss"
 	"github.com/influxdata/kapacitor/tick"
 	"github.com/influxdata/kapacitor/tick/stateful"
 	"github.com/influxdata/kapacitor/timer"
@@ -221,6 +222,11 @@ type TaskMaster struct {
 		StateChangesOnly() bool
 		Handler(servicenow.HandlerConfig, ...keyvalue.T) alert.Handler
 	}
+	ZenossService interface {
+		Global() bool
+		StateChangesOnly() bool
+		Handler(zenoss.HandlerConfig, ...keyvalue.T) alert.Handler
+	}
 
 	Commander command.Commander
 
@@ -319,6 +325,7 @@ func (tm *TaskMaster) New(id string) *TaskMaster {
 	n.SideloadService = tm.SideloadService
 	n.TeamsService = tm.TeamsService
 	n.ServiceNowService = tm.ServiceNowService
+	n.ZenossService = tm.ZenossService
 	return n
 }
 


Test patch:
diff --git a/client/v1/client_test.go b/client/v1/client_test.go
index e3627ae6..cdd1b8f9 100644
--- a/client/v1/client_test.go
+++ b/client/v1/client_test.go
@@ -47,6 +47,34 @@ func Test_ReportsErrors(t *testing.T) {
 				return err
 			},
 		},
+		{
+			name: "CreateUser",
+			fnc: func(c *client.Client) error {
+				_, err := c.CreateUser(client.CreateUserOptions{})
+				return err
+			},
+		},
+		{
+			name: "UpdateUser",
+			fnc: func(c *client.Client) error {
+				_, err := c.UpdateUser(c.UserLink(""), client.UpdateUserOptions{})
+				return err
+			},
+		},
+		{
+			name: "User",
+			fnc: func(c *client.Client) error {
+				_, err := c.User(c.UserLink(""))
+				return err
+			},
+		},
+		{
+			name: "DeleteUser",
+			fnc: func(c *client.Client) error {
+				err := c.DeleteUser(c.UserLink(""))
+				return err
+			},
+		},
 		{
 			name: "CreateTask",
 			fnc: func(c *client.Client) error {
@@ -3082,6 +3110,94 @@ func Test_LogLevel(t *testing.T) {
 	}
 }
 
+func Test_CreateUser(t *testing.T) {
+	s, c, err := newClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var user client.CreateUserOptions
+		body, _ := ioutil.ReadAll(r.Body)
+		err := json.Unmarshal(body, &user)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if r.URL.Path == "/kapacitor/v1/users" && r.Method == "POST" {
+			exp := client.CreateUserOptions{
+				Name: "username",
+				Type: client.AdminUser,
+				Permissions: []client.Permission{
+					client.AllPermissions,
+				},
+			}
+			if !reflect.DeepEqual(exp, user) {
+				w.WriteHeader(http.StatusBadRequest)
+				fmt.Fprintf(w, "unexpected CreateUser body: got:\n%v\nexp:\n%v\n", user, exp)
+			} else {
+				w.WriteHeader(http.StatusOK)
+				fmt.Fprint(w, `{"link": {"rel":"self", "href":"/kapacitor/v1/users/username"}, "name":"username"}`)
+			}
+		} else {
+			w.WriteHeader(http.StatusBadRequest)
+			fmt.Fprintf(w, "request: %v", r)
+		}
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	user, err := c.CreateUser(client.CreateUserOptions{
+		Name:        "username",
+		Type:        client.AdminUser,
+		Permissions: []client.Permission{client.AllPermissions},
+	})
+	if got, exp := string(user.Link.Href), "/kapacitor/v1/users/username"; got != exp {
+		t.Errorf("unexpected user link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Name, "username"; got != exp {
+		t.Errorf("unexpected username got %s exp %s", got, exp)
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func Test_User(t *testing.T) {
+	s, c, err := newClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/kapacitor/v1/users/bob" && r.Method == "GET" {
+			w.WriteHeader(http.StatusOK)
+			fmt.Fprintf(w, `{
+	"link": {"rel":"self", "href":"/kapacitor/v1/users/bob"},
+	"name": "bob",
+	"type":"normal",
+	"permissions":["api","write_points"]
+}`)
+		} else {
+			w.WriteHeader(http.StatusBadRequest)
+			fmt.Fprintf(w, "request: %v", r)
+		}
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	user, err := c.User(c.UserLink("bob"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	exp := client.User{
+		Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/users/bob"},
+		Name: "bob",
+		Type: client.NormalUser,
+		Permissions: []client.Permission{
+			client.APIPermission,
+			client.WritePointsPermission,
+		},
+	}
+	if !reflect.DeepEqual(exp, user) {
+		t.Errorf("unexpected user:\ngot:\n%v\nexp:\n%v", user, exp)
+	}
+}
+
 func Test_Bad_Creds(t *testing.T) {
 	testCases := []struct {
 		creds *client.Credentials
@@ -3185,3 +3301,234 @@ func Test_BearerAuthentication(t *testing.T) {
 		t.Errorf("unexpected version: got: %s exp: %s", got, exp)
 	}
 }
+
+func Test_DeleteUser(t *testing.T) {
+	s, c, err := newClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/kapacitor/v1/users/username" && r.Method == "DELETE" {
+			w.WriteHeader(http.StatusNoContent)
+		} else {
+			w.WriteHeader(http.StatusBadRequest)
+			fmt.Fprintf(w, "request: %v", r)
+		}
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	err = c.DeleteUser(c.UserLink("username"))
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func Test_ListUsers(t *testing.T) {
+	s, c, err := newClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/kapacitor/v1/users" && r.Method == "GET" &&
+			r.URL.Query().Get("pattern") == "" &&
+			r.URL.Query().Get("fields") == "" &&
+			r.URL.Query().Get("offset") == "0" &&
+			r.URL.Query().Get("limit") == "100" {
+			w.WriteHeader(http.StatusOK)
+			fmt.Fprintf(w, `{
+"users":[
+	{
+		"link": {"rel":"self", "href":"/kapacitor/v1/users/bob"},
+		"name": "bob",
+		"type":"admin"
+	},
+	{
+		"link": {"rel":"self", "href":"/kapacitor/v1/users/jim"},
+		"name": "jim",
+		"type":"normal",
+		"permissions":["none"]
+	}
+]}`)
+		} else {
+			w.WriteHeader(http.StatusBadRequest)
+			fmt.Fprintf(w, "request: %v", r)
+		}
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	users, err := c.ListUsers(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	exp := []client.User{
+		{
+			Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/users/bob"},
+			Name: "bob",
+			Type: client.AdminUser,
+		},
+		{
+			Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/users/jim"},
+			Name: "jim",
+			Type: client.NormalUser,
+			Permissions: []client.Permission{
+				client.NoPermissions,
+			},
+		},
+	}
+	if !reflect.DeepEqual(exp, users) {
+		t.Errorf("unexpected user list: got:\n%v\nexp:\n%v", users, exp)
+	}
+}
+
+func Test_ListUsers_Fields(t *testing.T) {
+	s, c, err := newClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/kapacitor/v1/users" && r.Method == "GET" &&
+			r.URL.Query().Get("pattern") == "" &&
+			len(r.URL.Query()["fields"]) == 1 &&
+			r.URL.Query()["fields"][0] == "type" &&
+			r.URL.Query().Get("offset") == "0" &&
+			r.URL.Query().Get("limit") == "100" {
+			w.WriteHeader(http.StatusOK)
+			fmt.Fprintf(w, `{
+"users":[
+	{
+		"link": {"rel":"self", "href":"/kapacitor/v1/users/bob"},
+		"name": "bob",
+		"type":"admin"
+	},
+	{
+		"link": {"rel":"self", "href":"/kapacitor/v1/users/jim"},
+		"name": "jim",
+		"type":"normal"
+	}
+]}`)
+		} else {
+			w.WriteHeader(http.StatusBadRequest)
+			fmt.Fprintf(w, "request: %v", r)
+		}
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	users, err := c.ListUsers(&client.ListUsersOptions{
+		Fields: []string{"type"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	exp := []client.User{
+		{
+			Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/users/bob"},
+			Name: "bob",
+			Type: client.AdminUser,
+		},
+		{
+			Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/users/jim"},
+			Name: "jim",
+			Type: client.NormalUser,
+		},
+	}
+	if !reflect.DeepEqual(exp, users) {
+		t.Errorf("unexpected user list: got:\n%v\nexp:\n%v", users, exp)
+	}
+}
+
+func Test_Permission_MarshalText(t *testing.T) {
+	for _, tc := range []struct {
+		permission client.Permission
+		text       string
+		err        error
+	}{
+		{
+			permission: client.NoPermissions,
+			text:       "none",
+		},
+		{
+			permission: client.APIPermission,
+			text:       "api",
+		},
+		{
+			permission: client.WritePointsPermission,
+			text:       "write_points",
+		},
+		{
+			permission: client.AllPermissions,
+			text:       "all",
+		},
+		{
+			permission: -1,
+			text:       "test",
+			err:        errors.New("unknown Permission -1"),
+		},
+	} {
+		if text, err := tc.permission.MarshalText(); err != nil {
+			if tc.err == nil {
+				t.Errorf("unexpected error for permission %v: %s", tc.permission, err)
+			} else if got, exp := err.Error(), tc.err.Error(); got != exp {
+				t.Errorf("unexpected permission error: got %s exp %s", got, exp)
+			} else {
+				p := client.Permission(-1)
+				if err := p.UnmarshalText([]byte(tc.text)); err == nil {
+					t.Errorf("expected error unmarshaling userType text: %s", tc.text)
+				}
+			}
+		} else if got, exp := string(text), tc.text; got != exp {
+			t.Errorf("unexpected permission text: got %s exp %s", got, exp)
+		} else {
+			p := client.Permission(-1)
+			if err := p.UnmarshalText(text); err != nil {
+				t.Errorf("unexpected error unmarshaling permission text: %s", string(text))
+			} else if got, exp := p, tc.permission; got != exp {
+				t.Errorf("unexpected permission: got %v exp %v", got, exp)
+			}
+		}
+	}
+}
+
+func Test_UserType_MarshalText(t *testing.T) {
+	for _, tc := range []struct {
+		userType client.UserType
+		text     string
+		err      error
+	}{
+		{
+			userType: client.InvalidUser,
+			text:     "",
+		},
+		{
+			userType: client.NormalUser,
+			text:     "normal",
+		},
+		{
+			userType: client.AdminUser,
+			text:     "admin",
+		},
+		{
+			userType: -1,
+			text:     "test",
+			err:      errors.New("unknown UserType -1"),
+		},
+	} {
+		if text, err := tc.userType.MarshalText(); err != nil {
+			if tc.err == nil {
+				t.Errorf("unexpected error for userType %v: %s", tc.userType, err)
+			} else if got, exp := err.Error(), tc.err.Error(); got != exp {
+				t.Errorf("unexpected userType error: got %s exp %s", got, exp)
+			} else {
+				ut := client.UserType(-1)
+				if err := ut.UnmarshalText([]byte(tc.text)); err == nil {
+					t.Errorf("expected error unmarshaling userType text: %s", tc.text)
+				}
+			}
+		} else if got, exp := string(text), tc.text; got != exp {
+			t.Errorf("unexpected userType text: got %s exp %s", got, exp)
+		} else {
+			ut := client.UserType(-1)
+			if err := ut.UnmarshalText(text); err != nil {
+				t.Errorf("unexpected error unmarshaling userType text: %s", string(text))
+			} else if got, exp := ut, tc.userType; got != exp {
+				t.Errorf("unexpected userType: got %v exp %v", got, exp)
+			}
+		}
+	}
+}
diff --git a/influxdb/client_test.go b/influxdb/client_test.go
index febe07b8..29cf054f 100644
--- a/influxdb/client_test.go
+++ b/influxdb/client_test.go
@@ -133,7 +133,7 @@ func TestClient_Concurrent_Use(t *testing.T) {
 
 		for i := 0; i < n; i++ {
 			if err = c.Write(bp); err != nil {
-				t.Fatalf("got error %v", err)
+				t.Errorf("got error %v", err)
 			}
 		}
 	}()
@@ -143,7 +143,7 @@ func TestClient_Concurrent_Use(t *testing.T) {
 		var q Query
 		for i := 0; i < n; i++ {
 			if _, err := c.Query(q); err != nil {
-				t.Fatalf("got error %v", err)
+				t.Errorf("got error %v", err)
 			}
 		}
 	}()
diff --git a/integrations/benchmark_test.go b/integrations/benchmark_test.go
index d2f58b9c..e3aba343 100644
--- a/integrations/benchmark_test.go
+++ b/integrations/benchmark_test.go
@@ -213,7 +213,7 @@ func Bench(b *testing.B, tasksCount, pointCount, expectedProcessedCount int, tic
 				responseRecorder := httptest.NewRecorder()
 				httpdService.Handler.ServeHTTP(responseRecorder, writeRequest)
 				if responseRecorder.Code != http.StatusNoContent {
-					b.Fatalf("failed to write test data %s", responseRecorder.Body.String())
+					b.Errorf("failed to write test data %s", responseRecorder.Body.String())
 				}
 			}(write.request, write.seeker)
 		}
diff --git a/integrations/streamer_test.go b/integrations/streamer_test.go
index b471793b..9d3e9fca 100644
--- a/integrations/streamer_test.go
+++ b/integrations/streamer_test.go
@@ -84,6 +84,8 @@ import (
 	"github.com/influxdata/kapacitor/services/telegram/telegramtest"
 	"github.com/influxdata/kapacitor/services/victorops"
 	"github.com/influxdata/kapacitor/services/victorops/victoropstest"
+	"github.com/influxdata/kapacitor/services/zenoss"
+	"github.com/influxdata/kapacitor/services/zenoss/zenosstest"
 	"github.com/influxdata/kapacitor/udf"
 	"github.com/influxdata/kapacitor/udf/agent"
 	udf_test "github.com/influxdata/kapacitor/udf/test"
@@ -7253,7 +7255,7 @@ stream
 		}
 
 		if got, exp := len(init.Options), 2; got != exp {
-			t.Fatalf("unexpected number of options in init request, got %d exp %d", got, exp)
+			t.Errorf("unexpected number of options in init request, got %d exp %d", got, exp)
 		}
 		for i, opt := range init.Options {
 			exp := &agent.Option{}
@@ -8810,7 +8812,7 @@ stream
 	exp := []interface{}{
 		kafkatest.Message{
 			Topic:     "testTopic",
-			Partition: 1,
+			Partition: 2,
 			Offset:    0,
 			Key:       "kapacitor/cpu/serverA",
 			Message:   "kapacitor/cpu/serverA is CRITICAL",
@@ -8852,6 +8854,64 @@ stream
 	}
 }
 
+func TestStream_AlertKafka_Partitioning(t *testing.T) {
+	ts, err := kafkatest.NewServer()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ts.Close()
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+	|alert()
+		.id('{{ index .Tags "host" }}')
+		.crit(lambda: TRUE)
+		.kafka()
+		.cluster('default')
+		.kafkaTopic('testTopic')
+		.template('{{.Message}}')
+`
+
+	tmInit := func(tm *kapacitor.TaskMaster) {
+		configs := kafka.Configs{{
+			Enabled:   true,
+			ID:        "default",
+			Brokers:   []string{ts.Addr.String()},
+			BatchSize: 1,
+		}}
+		d := diagService.NewKafkaHandler().WithContext(keyvalue.KV("test", "kafka"))
+		tm.KafkaService = kafka.NewService(configs, d)
+	}
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	// Wait for kakfa messages to be written
+	time.Sleep(time.Second)
+
+	ts.Close()
+	msgs, err := ts.Messages()
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := make(map[string]map[int32]struct{})
+	for _, m := range msgs {
+		// Record the partitions used per message key.
+		partitions, ok := got[m.Key]
+		if !ok {
+			got[m.Key] = make(map[int32]struct{})
+			partitions = got[m.Key]
+		}
+		partitions[m.Partition] = struct{}{}
+	}
+
+	for msg, partitions := range got {
+		if len(partitions) > 1 {
+			t.Fatalf("messages for %q were sent to %d partitions, expected only 1 partition", msg, len(partitions))
+		}
+	}
+}
+
 func TestStream_AlertTelegram(t *testing.T) {
 	ts := telegramtest.NewServer()
 	defer ts.Close()
@@ -11149,6 +11209,160 @@ stream
 	}
 }
 
+func TestStream_AlertZenoss(t *testing.T) {
+	ts := zenosstest.NewServer()
+	defer ts.Close()
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host', 'type')
+	|window()
+		.period(10s)
+		.every(10s)
+	|mean('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.info(lambda: "mean" > 15.0)
+		.warn(lambda: "mean" > 50.0)
+		.crit(lambda: "mean" > 90.0)
+		.zenoss()
+			.device('#DEVICE001')
+			.component('CPU')
+			.eventClass('/App')
+`
+
+	tmInit := func(tm *kapacitor.TaskMaster) {
+
+		c := zenoss.NewConfig()
+		c.Enabled = true
+		c.URL = ts.URL + "/zport/dmd/evconsole_router"
+		svc := zenoss.NewService(c, diagService.NewZenossHandler())
+		tm.ZenossService = svc
+	}
+
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	exp := []interface{}{
+		zenosstest.Request{
+			URL: "/zport/dmd/evconsole_router",
+			Event: zenoss.Event{
+				Action: "EventsRouter",
+				Method: "add_event",
+				Data: []map[string]interface{}{
+					{
+						"summary":    "kapacitor/cpu/serverA is CRITICAL",
+						"device":     "#DEVICE001",
+						"component":  "CPU",
+						"severity":   "Critical",
+						"evclasskey": "",
+						"evclass":    "/App",
+					},
+				},
+				Type: "rpc",
+				TID:  1,
+			},
+		},
+	}
+
+	ts.Close()
+	var got []interface{}
+	for _, g := range ts.Requests() {
+		got = append(got, g)
+	}
+
+	if err := compareListIgnoreOrder(got, exp, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestStream_AlertZenoss_Custom(t *testing.T) {
+	ts := zenosstest.NewServer()
+	defer ts.Close()
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host', 'type')
+	|window()
+		.period(10s)
+		.every(10s)
+	|mean('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.info(lambda: "mean" > 15.0)
+		.warn(lambda: "mean" > 50.0)
+		.crit(lambda: "mean" > 90.0)
+		.details('{{.}}')
+		.zenoss()
+			.action('ScriptsRouter')
+			.method('kapa_handler')
+			.eventClass('/App')
+			.message('This is message for alert {{ .ID }}')
+			.collector('{{ index .Tags "host" }}')
+			.customField('data', '{"id":"{{.ID}}","message":"{{.Message}}","time":"{{.Time}}","duration":"{{ .Duration}}","level":"{{.Level}}","recoverable":{{.Recoverable}}}')
+			.customField('ticks', 33)
+`
+	tmInit := func(tm *kapacitor.TaskMaster) {
+
+		c := zenoss.NewConfig()
+		c.Enabled = true
+		c.URL = ts.URL + "/zport/dmd/evconsole_router"
+		c.SeverityMap = zenoss.SeverityMap{OK: 0, Info: 2, Warning: 3, Critical: 5}
+		svc := zenoss.NewService(c, diagService.NewZenossHandler())
+		tm.ZenossService = svc
+	}
+
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	exp := []interface{}{
+		zenosstest.Request{
+			URL: "/zport/dmd/evconsole_router",
+			Event: zenoss.Event{
+				Action: "ScriptsRouter",
+				Method: "kapa_handler",
+				Data: []map[string]interface{}{
+					{
+						"summary":    "kapacitor/cpu/serverA is CRITICAL",
+						"device":     "",
+						"component":  "",
+						"severity":   float64(5),
+						"evclasskey": "",
+						"evclass":    "/App",
+						"collector":  "serverA",
+						"message":    "This is message for alert kapacitor/cpu/serverA",
+						"data": map[string]interface{}{
+							"id":          "kapacitor/cpu/serverA",
+							"level":       "CRITICAL",
+							"message":     "kapacitor/cpu/serverA is CRITICAL",
+							"time":        "1971-01-01 00:00:10 +0000 UTC",
+							"duration":    "0s",
+							"recoverable": true,
+						},
+						"ticks": float64(33),
+					},
+				},
+				Type: "rpc",
+				TID:  1,
+			},
+		},
+	}
+
+	ts.Close()
+	var got []interface{}
+	for _, g := range ts.Requests() {
+		got = append(got, g)
+	}
+
+	if err := compareListIgnoreOrder(got, exp, nil); err != nil {
+		t.Error(err)
+	}
+}
+
 func TestStream_LambdaNow(t *testing.T) {
 	var script = `
 stream
diff --git a/pipeline/alert_test.go b/pipeline/alert_test.go
index fba170ca..1f5d761b 100644
--- a/pipeline/alert_test.go
+++ b/pipeline/alert_test.go
@@ -84,7 +84,160 @@ func TestAlertNode_MarshalJSON(t *testing.T) {
     "snmpTrap": null,
     "kafka": null,
     "teams": null,
-    "serviceNow": null
+    "serviceNow": null,
+    "zenoss": null
+}`,
+		},
+		{
+			name: "marshal kafka with partition-by-id enabled",
+			node: AlertNode{
+				AlertNodeData: &AlertNodeData{
+					KafkaHandlers: []*KafkaHandler{
+						{
+							Cluster:                "my-cluster",
+							KafkaTopic:             "my-topic",
+							IsDisablePartitionById: false,
+							PartitionHashAlgorithm: "murmur2",
+						},
+					},
+				},
+			},
+			want: `{
+    "typeOf": "alert",
+    "id": "0",
+    "category": "",
+    "topic": "",
+    "alertId": "",
+    "message": "",
+    "details": "",
+    "info": null,
+    "warn": null,
+    "crit": null,
+    "infoReset": null,
+    "warnReset": null,
+    "critReset": null,
+    "useFlapping": false,
+    "flapLow": 0,
+    "flapHigh": 0,
+    "history": 0,
+    "levelTag": "",
+    "levelField": "",
+    "messageField": "",
+    "durationField": "",
+    "idTag": "",
+    "idField": "",
+    "all": false,
+    "noRecoveries": false,
+    "stateChangesOnly": false,
+    "stateChangesOnlyDuration": 0,
+    "inhibitors": null,
+    "post": null,
+    "tcp": null,
+    "email": null,
+    "exec": null,
+    "log": null,
+    "victorOps": null,
+    "pagerDuty": null,
+    "pagerDuty2": null,
+    "pushover": null,
+    "sensu": null,
+    "slack": null,
+    "discord": null,
+    "bigPanda": null,
+    "telegram": null,
+    "hipChat": null,
+    "alerta": null,
+    "opsGenie": null,
+    "opsGenie2": null,
+    "talk": null,
+    "mqtt": null,
+    "snmpTrap": null,
+    "kafka": [
+        {
+            "cluster": "my-cluster",
+            "kafka-topic": "my-topic",
+            "disable-partition-by-id": false,
+            "partition-hash-algorithm": "murmur2"
+        }
+    ],
+    "teams": null,
+    "serviceNow": null,
+    "zenoss": null
+}`,
+		},
+		{
+			name: "marshal kafka with partition-by-id disabled",
+			node: AlertNode{
+				AlertNodeData: &AlertNodeData{
+					KafkaHandlers: []*KafkaHandler{
+						{
+							Cluster:                "my-cluster",
+							KafkaTopic:             "my-topic",
+							IsDisablePartitionById: true,
+						},
+					},
+				},
+			},
+			want: `{
+    "typeOf": "alert",
+    "id": "0",
+    "category": "",
+    "topic": "",
+    "alertId": "",
+    "message": "",
+    "details": "",
+    "info": null,
+    "warn": null,
+    "crit": null,
+    "infoReset": null,
+    "warnReset": null,
+    "critReset": null,
+    "useFlapping": false,
+    "flapLow": 0,
+    "flapHigh": 0,
+    "history": 0,
+    "levelTag": "",
+    "levelField": "",
+    "messageField": "",
+    "durationField": "",
+    "idTag": "",
+    "idField": "",
+    "all": false,
+    "noRecoveries": false,
+    "stateChangesOnly": false,
+    "stateChangesOnlyDuration": 0,
+    "inhibitors": null,
+    "post": null,
+    "tcp": null,
+    "email": null,
+    "exec": null,
+    "log": null,
+    "victorOps": null,
+    "pagerDuty": null,
+    "pagerDuty2": null,
+    "pushover": null,
+    "sensu": null,
+    "slack": null,
+    "discord": null,
+    "bigPanda": null,
+    "telegram": null,
+    "hipChat": null,
+    "alerta": null,
+    "opsGenie": null,
+    "opsGenie2": null,
+    "talk": null,
+    "mqtt": null,
+    "snmpTrap": null,
+    "kafka": [
+        {
+            "cluster": "my-cluster",
+            "kafka-topic": "my-topic",
+            "disable-partition-by-id": true
+        }
+    ],
+    "teams": null,
+    "serviceNow": null,
+    "zenoss": null
 }`,
 		},
 	}
diff --git a/pipeline/json_test.go b/pipeline/json_test.go
index fe0dbd4d..479a008d 100644
--- a/pipeline/json_test.go
+++ b/pipeline/json_test.go
@@ -263,7 +263,8 @@ func TestPipeline_MarshalJSON(t *testing.T) {
             "snmpTrap": null,
             "kafka": null,
             "teams": null,
-            "serviceNow": null
+            "serviceNow": null,
+            "zenoss": null
         },
         {
             "typeOf": "httpOut",
diff --git a/pipeline/tick/alert_test.go b/pipeline/tick/alert_test.go
index 121d6111..1802fde0 100644
--- a/pipeline/tick/alert_test.go
+++ b/pipeline/tick/alert_test.go
@@ -588,12 +588,13 @@ func TestAlertHipchat(t *testing.T) {
 	PipelineTickTestHelper(t, pipe, want)
 }
 
-func TestAlertKafka(t *testing.T) {
+func TestAlertKafka_PartitionById(t *testing.T) {
 	pipe, _, from := StreamFrom()
 	handler := from.Alert().Kafka()
 	handler.Cluster = "default"
 	handler.KafkaTopic = "test"
 	handler.Template = "tmpl"
+	handler.PartitionHashAlgorithm = "murmur2"
 
 	want := `stream
     |from()
@@ -605,6 +606,31 @@ func TestAlertKafka(t *testing.T) {
         .kafka()
         .cluster('default')
         .kafkaTopic('test')
+        .partitionHashAlgorithm('murmur2')
+        .template('tmpl')
+`
+	PipelineTickTestHelper(t, pipe, want)
+}
+
+func TestAlertKafka_NoPartitionById(t *testing.T) {
+	pipe, _, from := StreamFrom()
+	handler := from.Alert().Kafka()
+	handler.Cluster = "default"
+	handler.KafkaTopic = "test"
+	handler.Template = "tmpl"
+	handler.IsDisablePartitionById = true
+
+	want := `stream
+    |from()
+    |alert()
+        .id('{{ .Name }}:{{ .Group }}')
+        .message('{{ .ID }} is {{ .Level }}')
+        .details('{{ json . }}')
+        .history(21)
+        .kafka()
+        .cluster('default')
+        .kafkaTopic('test')
+        .disablePartitionById()
         .template('tmpl')
 `
 	PipelineTickTestHelper(t, pipe, want)
@@ -764,3 +790,29 @@ func TestAlertSNMP(t *testing.T) {
 `
 	PipelineTickTestHelper(t, pipe, want)
 }
+
+func TestAlertZenoss(t *testing.T) {
+	pipe, _, from := StreamFrom()
+	handler := from.Alert().Zenoss()
+	handler.Action = "custom_add_event"
+	handler.Summary = "server alert"
+	handler.Device = "server-A"
+	handler.Component = "CPU"
+	handler.EventClass = "/App"
+
+	want := `stream
+    |from()
+    |alert()
+        .id('{{ .Name }}:{{ .Group }}')
+        .message('{{ .ID }} is {{ .Level }}')
+        .details('{{ json . }}')
+        .history(21)
+        .zenoss()
+        .action('custom_add_event')
+        .summary('server alert')
+        .device('server-A')
+        .component('CPU')
+        .eventClass('/App')
+`
+	PipelineTickTestHelper(t, pipe, want)
+}
diff --git a/server/server_helper_test.go b/server/server_helper_test.go
index c2ff1396..610ef4df 100644
--- a/server/server_helper_test.go
+++ b/server/server_helper_test.go
@@ -203,6 +203,37 @@ func (s *Server) MustWrite(db, rp, body string, params url.Values) string {
 	return results
 }
 
+func (s *Server) PingAsUser(username, password string) error {
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: username,
+			Password: password,
+		},
+	})
+	if err != nil {
+		return err
+	}
+	_, _, err = cli.Ping()
+	return err
+}
+
+func (s *Server) PingWithToken(token string) error {
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method: client.BearerAuthentication,
+			Token:  token,
+		},
+	})
+	if err != nil {
+		return err
+	}
+	_, _, err = cli.Ping()
+	return err
+}
+
 type stats struct {
 	NumEnabledTasks int `json:"num_enabled_tasks"`
 	NumTasks        int `json:"num_tasks"`
diff --git a/server/server_test.go b/server/server_test.go
index 7701d455..eaa7ee45 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -10,6 +10,7 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"net/http"
+	"net/http/httptest"
 	"net/mail"
 	"net/url"
 	"os"
@@ -39,6 +40,8 @@ import (
 	"github.com/influxdata/kapacitor/server"
 	"github.com/influxdata/kapacitor/services/alert/alerttest"
 	"github.com/influxdata/kapacitor/services/alerta/alertatest"
+	"github.com/influxdata/kapacitor/services/auth"
+	"github.com/influxdata/kapacitor/services/auth/meta"
 	"github.com/influxdata/kapacitor/services/bigpanda/bigpandatest"
 	"github.com/influxdata/kapacitor/services/discord/discordtest"
 	"github.com/influxdata/kapacitor/services/hipchat/hipchattest"
@@ -73,8 +76,11 @@ import (
 	"github.com/influxdata/kapacitor/services/udf"
 	"github.com/influxdata/kapacitor/services/victorops"
 	"github.com/influxdata/kapacitor/services/victorops/victoropstest"
+	"github.com/influxdata/kapacitor/services/zenoss"
+	"github.com/influxdata/kapacitor/services/zenoss/zenosstest"
 	"github.com/k-sone/snmpgo"
 	"github.com/pkg/errors"
+	"golang.org/x/crypto/bcrypt"
 )
 
 var udfDir string
@@ -84,6 +90,117 @@ func init() {
 	udfDir = filepath.Clean(filepath.Join(dir, "../udf"))
 }
 
+func mustHash(hash []byte, err error) string {
+	if err != nil {
+		panic(err)
+	}
+	return string(hash)
+}
+
+func TestService_Authenticate(t *testing.T) {
+	type Users struct {
+		Users []meta.User `json:"users"`
+	}
+	newUsers := func(u ...meta.User) Users {
+		return Users{
+			Users: u,
+		}
+	}
+	cases := []struct {
+		name             string
+		userResult       Users
+		authorizedStatus int
+		shouldErr        bool
+		userName         string
+		password         string
+	}{{
+		name: "ok",
+		userResult: newUsers(meta.User{
+			Name:        "fred",
+			Hash:        mustHash(bcrypt.GenerateFromPassword([]byte(`fred_pw`), auth.NewEnabledConfig().BcryptCost)),
+			Permissions: map[string][]meta.Permission{"": {meta.KapacitorAPIPermission}},
+		}),
+		authorizedStatus: http.StatusOK,
+		shouldErr:        false,
+		userName:         `fred`,
+		password:         `fred_pw`,
+	},
+		{
+			name: "ldap badpass",
+			userResult: newUsers(meta.User{
+				Name:        "fred2",
+				Hash:        "",
+				Permissions: map[string][]meta.Permission{"": {meta.KapacitorAPIPermission}},
+			}),
+			authorizedStatus: http.StatusUnauthorized,
+			shouldErr:        true,
+			userName:         `fred2`,
+			password:         `badpass`,
+		},
+		{
+			name: "ldap ok",
+			userResult: newUsers(meta.User{
+				Name:        "fred3",
+				Hash:        "",
+				Permissions: map[string][]meta.Permission{"": {meta.KapacitorAPIPermission}},
+			}),
+			authorizedStatus: http.StatusOK,
+			shouldErr:        false,
+			userName:         `fred3`,
+			password:         `fred_pw`,
+		},
+		{
+			name:             "user dos not exist",
+			userResult:       newUsers(meta.User{}),
+			authorizedStatus: http.StatusNotFound,
+			shouldErr:        true,
+			userName:         `fred_not_exists`,
+			password:         `fred_pw`,
+		},
+	}
+	for _, testCase := range cases {
+		testCase := testCase
+		t.Run(testCase.name, func(t *testing.T) {
+			t.Parallel()
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.URL.Path {
+				case "/user":
+					if err := json.NewEncoder(w).Encode(testCase.userResult); err != nil {
+						t.Error("bad encoding for testCase.userResult")
+					}
+					//w.Write(testCase.userResult)
+				case "/authorized":
+					w.WriteHeader(testCase.authorizedStatus)
+				default:
+					t.Errorf("bad path %s", r.URL.Path)
+				}
+			}))
+			defer srv.Close()
+			c := NewConfig()
+			c.Auth = auth.NewEnabledConfig()
+			kserver := OpenServer(c)
+			url, err := url.Parse(srv.URL)
+			if err != nil {
+				t.Fatal(err)
+			}
+			kserver.Config.Auth.MetaAddr = url.Host
+			kserver.Restart()
+			defer kserver.Close()
+
+			_, err = kserver.AuthService.Authenticate(testCase.userName, testCase.password)
+			if (err != nil) != testCase.shouldErr {
+				if err == nil {
+					t.Log("expected an error but it didn't")
+				}
+				if err != nil {
+					t.Logf("expected no error but got %s", err.Error())
+				}
+				t.FailNow()
+			}
+		})
+	}
+}
+
 func TestServer_Ping(t *testing.T) {
 	s, cli := OpenDefaultServer()
 	defer s.Close()
@@ -305,6 +422,58 @@ func TestServer_Authenticate_Bearer(t *testing.T) {
 	}
 }
 
+func TestServer_CreateUser(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	username := "bob"
+	utype := client.NormalUser
+	permissions := []client.Permission{
+		client.APIPermission,
+	}
+	user, err := cli.CreateUser(client.CreateUserOptions{
+		Name:        username,
+		Password:    "hunter2",
+		Type:        utype,
+		Permissions: permissions,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if !reflect.DeepEqual(user.Permissions, permissions) {
+		t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+	}
+
+	user, err = cli.User(user.Link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if !reflect.DeepEqual(user.Permissions, permissions) {
+		t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+	}
+}
 func TestServer_CreateTask(t *testing.T) {
 	s, cli := OpenDefaultServer()
 	defer s.Close()
@@ -8564,6 +8733,119 @@ func TestServer_UpdateConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			section: "zenoss",
+			setDefaults: func(c *server.Config) {
+				c.Zenoss.URL = "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router"
+			},
+			expDefaultSection: client.ConfigSection{
+				Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss"},
+				Elements: []client.ConfigElement{{
+					Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss/"},
+					Options: map[string]interface{}{
+						"enabled":            false,
+						"global":             false,
+						"state-changes-only": false,
+						"url":                "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router",
+						"username":           "",
+						"password":           false,
+						"action":             "EventsRouter",
+						"method":             "add_event",
+						"type":               "rpc",
+						"tid":                float64(1),
+						"severity-map": map[string]interface{}{
+							"ok": "Clear", "info": "Info", "warning": "Warning", "critical": "Critical",
+						},
+					},
+					Redacted: []string{
+						"password",
+					},
+				}},
+			},
+			expDefaultElement: client.ConfigElement{
+				Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss/"},
+				Options: map[string]interface{}{
+					"enabled":            false,
+					"global":             false,
+					"state-changes-only": false,
+					"url":                "https://tenant.zenoss.io:8080/zport/dmd/evconsole_router",
+					"username":           "",
+					"password":           false,
+					"action":             "EventsRouter",
+					"method":             "add_event",
+					"type":               "rpc",
+					"tid":                float64(1),
+					"severity-map": map[string]interface{}{
+						"ok": "Clear", "info": "Info", "warning": "Warning", "critical": "Critical",
+					},
+				},
+				Redacted: []string{
+					"password",
+				},
+			},
+			updates: []updateAction{
+				{
+					updateAction: client.ConfigUpdateAction{
+						Set: map[string]interface{}{
+							"enabled":  true,
+							"url":      "https://dev12345.zenoss.io:8080/zport/dmd/evconsole_router",
+							"username": "dev",
+							"password": "12345",
+							"action":   "ScriptsRouter",
+							"method":   "kapa_handler",
+							"severity-map": zenoss.SeverityMap{
+								OK: 0, Info: 2, Warning: 3, Critical: 5,
+							},
+						},
+					},
+					expSection: client.ConfigSection{
+						Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss"},
+						Elements: []client.ConfigElement{{
+							Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss/"},
+							Options: map[string]interface{}{
+								"enabled":            true,
+								"global":             false,
+								"state-changes-only": false,
+								"url":                "https://dev12345.zenoss.io:8080/zport/dmd/evconsole_router",
+								"username":           "dev",
+								"password":           true,
+								"action":             "ScriptsRouter",
+								"method":             "kapa_handler",
+								"type":               "rpc",
+								"tid":                float64(1),
+								"severity-map": map[string]interface{}{
+									"ok": float64(0), "info": float64(2), "warning": float64(3), "critical": float64(5),
+								},
+							},
+							Redacted: []string{
+								"password",
+							},
+						}},
+					},
+					expElement: client.ConfigElement{
+						Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/config/zenoss/"},
+						Options: map[string]interface{}{
+							"enabled":            true,
+							"global":             false,
+							"state-changes-only": false,
+							"url":                "https://dev12345.zenoss.io:8080/zport/dmd/evconsole_router",
+							"username":           "dev",
+							"password":           true,
+							"action":             "ScriptsRouter",
+							"method":             "kapa_handler",
+							"type":               "rpc",
+							"tid":                float64(1),
+							"severity-map": map[string]interface{}{
+								"ok": float64(0), "info": float64(2), "warning": float64(3), "critical": float64(5),
+							},
+						},
+						Redacted: []string{
+							"password",
+						},
+					},
+				},
+			},
+		},
 	}
 
 	compareElements := func(got, exp client.ConfigElement) error {
@@ -9060,6 +9342,26 @@ func TestServer_ListServiceTests(t *testing.T) {
 					"entityID":    "testEntityID",
 				},
 			},
+			{
+				Link: client.Link{Relation: client.Self, Href: "/kapacitor/v1/service-tests/zenoss"},
+				Name: "zenoss",
+				Options: client.ServiceTestOptions{
+					"alert_id":        "1001",
+					"level":           "CRITICAL",
+					"message":         "test zenoss message",
+					"action":          "EventsRouter",
+					"method":          "add_event",
+					"type":            "rpc",
+					"tid":             float64(1),
+					"summary":         "",
+					"device":          "",
+					"component":       "",
+					"event_class_key": "",
+					"event_class":     "",
+					"collector":       "",
+					"custom_fields":   map[string]interface{}{},
+				},
+			},
 		},
 	}
 	if got, exp := serviceTests.Link.Href, expServiceTests.Link.Href; got != exp {
@@ -9384,6 +9686,14 @@ func TestServer_DoServiceTest(t *testing.T) {
 				Message: "service is not enabled",
 			},
 		},
+		{
+			service: "zenoss",
+			options: client.ServiceTestOptions{},
+			exp: client.ServiceTestResult{
+				Success: false,
+				Message: "service is not enabled",
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -9866,7 +10176,7 @@ func TestServer_AlertHandlers(t *testing.T) {
 				}
 				exp := []kafkatest.Message{{
 					Topic:     "test",
-					Partition: 1,
+					Partition: 3,
 					Offset:    0,
 					Key:       "id",
 					Message:   string(adJSON) + "\n",
@@ -10729,10 +11039,58 @@ func TestServer_AlertHandlers(t *testing.T) {
 				return nil
 			},
 		},
+		{
+			handler: client.TopicHandler{
+				Kind: "zenoss",
+				Options: map[string]interface{}{
+					"action": "EventsRouter",
+					"method": "add_event",
+					"type":   "rpc",
+					"TID":    1,
+				},
+			},
+			setup: func(c *server.Config, ha *client.TopicHandler) (context.Context, error) {
+				ts := zenosstest.NewServer()
+				ctxt := context.WithValue(context.Background(), "server", ts)
+
+				c.Zenoss.Enabled = true
+				c.Zenoss.URL = ts.URL + "/zport/dmd/evconsole_router"
+				return ctxt, nil
+			},
+			result: func(ctxt context.Context) error {
+				ts := ctxt.Value("server").(*zenosstest.Server)
+				ts.Close()
+				exp := []zenosstest.Request{{
+					URL: "/zport/dmd/evconsole_router",
+					Event: zenoss.Event{
+						Action: "EventsRouter",
+						Method: "add_event",
+						Data: []map[string]interface{}{
+							{
+								"component":  "",
+								"device":     "",
+								"evclass":    "",
+								"evclasskey": "",
+								"severity":   "Critical",
+								"summary":    "message",
+							},
+						},
+						Type: "rpc",
+						TID:  1,
+					},
+				}}
+				got := ts.Requests()
+				if !reflect.DeepEqual(exp, got) {
+					return fmt.Errorf("unexpected zenoss request:\nexp\n%+v\ngot\n%+v\n", exp, got)
+				}
+				return nil
+			},
+		},
 	}
 	for i, tc := range testCases {
 		t.Run(fmt.Sprintf("%s-%d", tc.handler.Kind, i), func(t *testing.T) {
 			kind := tc.handler.Kind
+
 			// Create default config
 			c := NewConfig()
 			var ctxt context.Context
@@ -12876,3 +13234,678 @@ func compareListIgnoreOrder(got, exp []interface{}, cmpF func(got, exp interface
 	}
 	return nil
 }
+
+func TestServer_Authenticate_Fail_Enterprise(t *testing.T) {
+	t.Parallel()
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	s := OpenServer(conf)
+	defer s.Close()
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = cli.Ping()
+	if err == nil {
+		t.Error("expected authentication error")
+	} else if exp, got := "unable to parse authentication credentials", err.Error(); got != exp {
+		t.Errorf("unexpected error message: got %q exp %q", got, exp)
+	}
+}
+
+func TestServer_Authenticate_User_Enterprise(t *testing.T) {
+	t.Parallel()
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	s := OpenServer(conf)
+	defer s.Close()
+	s.AuthService.(*auth.Service).CreateUser("root", "kapacitor", true, nil)
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "root",
+			Password: "kapacitor",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     "bob",
+		Password: "secret",
+		Type:     client.AdminUser,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PingAsUser("bob", "secret"); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestServer_Authenticate_Bearer_Fail_Enterprise(t *testing.T) {
+	t.Parallel()
+	secret := "secret"
+	// Create a new token object, specifying signing method and the claims
+	// you would like it to contain.
+	token := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
+		"username": "bob",
+		"exp":      time.Now().Add(10 * time.Second).Unix(),
+	})
+
+	// Sign and get the complete encoded token as a string using the secret
+	tokenString, err := token.SignedString([]byte(secret))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	// Use a different secret so the token is invalid
+	conf.HTTP.SharedSecret = secret + "extra secret"
+	s := OpenServer(conf)
+	defer s.Close()
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method: client.BearerAuthentication,
+			Token:  tokenString,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = cli.Ping()
+	if err == nil {
+		t.Error("expected authentication error")
+	} else if exp, got := "invalid token: signature is invalid", err.Error(); got != exp {
+		t.Errorf("unexpected error message: got %q exp %q", got, exp)
+	}
+}
+
+func TestServer_Authenticate_Bearer_Expired_Enteprise(t *testing.T) {
+	t.Parallel()
+	secret := "secret"
+	// Create a new token object, specifying signing method and the claims
+	// you would like it to contain.
+	token := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
+		"username": "bob",
+		"exp":      time.Now().Add(-10 * time.Second).Unix(),
+	})
+
+	// Sign and get the complete encoded token as a string using the secret
+	tokenString, err := token.SignedString([]byte(secret))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conf := NewConfig()
+	conf.HTTP.AuthEnabled = true
+	conf.HTTP.SharedSecret = secret
+	s := OpenServer(conf)
+	defer s.Close()
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method: client.BearerAuthentication,
+			Token:  tokenString,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = cli.Ping()
+	if err == nil {
+		t.Error("expected authentication error")
+	} else if exp, got := "invalid token: Token is expired", err.Error(); got != exp {
+		t.Errorf("unexpected error message: got %q exp %q", got, exp)
+	}
+}
+
+func TestServer_Authenticate_Bearer_Enterprise(t *testing.T) {
+	t.Parallel()
+	secret := "secret"
+	// Create a new token object, specifying signing method and the claims
+	// you would like it to contain.
+	token := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
+		"username": "bob",
+		"exp":      time.Now().Add(time.Minute).Unix(),
+	})
+
+	// Sign and get the complete encoded token as a string using the secret
+	tokenString, err := token.SignedString([]byte(secret))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	conf.HTTP.SharedSecret = secret
+	s := OpenServer(conf)
+	defer s.Close()
+	s.AuthService.(*auth.Service).CreateUser("root", "kapacitor", true, nil)
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "root",
+			Password: "kapacitor",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     "bob",
+		Password: "secret",
+		Type:     client.AdminUser,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PingWithToken(tokenString); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestServer_Authenticate_Bearer_PermissionDenied(t *testing.T) {
+	t.Parallel()
+	secret := "secret"
+	// Create a new token object, specifying signing method and the claims
+	// you would like it to contain.
+	token := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
+		"username": "bob",
+		"exp":      time.Now().Add(time.Minute).Unix(),
+	})
+
+	// Sign and get the complete encoded token as a string using the secret
+	tokenString, err := token.SignedString([]byte(secret))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	conf.HTTP.SharedSecret = secret
+	s := OpenServer(conf)
+	defer s.Close()
+	s.AuthService.(*auth.Service).CreateUser("root", "kapacitor", true, nil)
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "root",
+			Password: "kapacitor",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     "bob",
+		Password: "secret",
+		Type:     client.NormalUser,
+		Permissions: []client.Permission{
+			client.WritePointsPermission,
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if bobsCLI, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method: client.BearerAuthentication,
+			Token:  tokenString,
+		},
+	}); err != nil {
+		t.Fatal(err)
+	} else {
+		// Ping is allowed
+		if _, version, err := bobsCLI.Ping(); err != nil {
+			t.Fatal(err)
+		} else if version != "testServer" {
+			t.Fatal("unexpected version", version)
+		}
+		// Delete user is not allowed
+		if err := bobsCLI.DeleteUser(bobsCLI.UserLink("root")); err == nil {
+			t.Fatal("expected permission denied error")
+		} else if got, exp := err.Error(), `user bob does not have "delete" privilege for API endpoint "/kapacitor/v1/users/root"`; got != exp {
+			t.Errorf("unexpected error message: got %s exp %s", got, exp)
+		}
+	}
+}
+
+func TestServer_Authenticate_User_PermissionDenied(t *testing.T) {
+	t.Parallel()
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	s := OpenServer(conf)
+	defer s.Close()
+	s.AuthService.(*auth.Service).CreateUser("root", "kapacitor", true, nil)
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "root",
+			Password: "kapacitor",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     "bob",
+		Password: "secret",
+		Type:     client.NormalUser,
+		Permissions: []client.Permission{
+			client.WritePointsPermission,
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if bobsCLI, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "bob",
+			Password: "secret",
+		},
+	}); err != nil {
+		t.Fatal(err)
+	} else {
+		// Ping is allowed
+		if _, version, err := bobsCLI.Ping(); err != nil {
+			t.Fatal(err)
+		} else if version != "testServer" {
+			t.Fatal("unexpected version", version)
+		}
+		// Delete user is not allowed
+		if err := bobsCLI.DeleteUser(bobsCLI.UserLink("root")); err == nil {
+			t.Fatal("expected permission denied error")
+		} else if got, exp := err.Error(), `user bob does not have "delete" privilege for API endpoint "/kapacitor/v1/users/root"`; got != exp {
+			t.Errorf("unexpected error message: got %s exp %s", got, exp)
+		}
+	}
+}
+
+func TestServer_Authenticate_User_Multiple(t *testing.T) {
+	t.Parallel()
+	conf := NewConfig()
+	conf.Auth = auth.NewEnabledConfig()
+	conf.HTTP.AuthEnabled = true
+	s := OpenServer(conf)
+	defer s.Close()
+	s.AuthService.(*auth.Service).CreateUser("root", "kapacitor", true, nil)
+	cli, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "root",
+			Password: "kapacitor",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     "bob",
+		Password: "secret",
+		Type:     client.NormalUser,
+		Permissions: []client.Permission{
+			client.WritePointsPermission,
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if bobsCLI, err := client.New(client.Config{
+		URL: s.URL(),
+		Credentials: &client.Credentials{
+			Method:   client.UserAuthentication,
+			Username: "bob",
+			Password: "secret",
+		},
+	}); err != nil {
+		t.Fatal(err)
+	} else {
+		count := 10
+		for i := 0; i < count; i++ {
+			// Ping is allowed
+			if _, version, err := bobsCLI.Ping(); err != nil {
+				t.Fatal(err)
+			} else if version != "testServer" {
+				t.Fatal("unexpected version", version)
+			}
+			// Delete user is not allowed
+			if err := bobsCLI.DeleteUser(bobsCLI.UserLink("root")); err == nil {
+				t.Fatal("expected permission denied error")
+			} else if got, exp := err.Error(), `user bob does not have "delete" privilege for API endpoint "/kapacitor/v1/users/root"`; got != exp {
+				t.Errorf("unexpected error message: got %s exp %s", got, exp)
+			}
+		}
+	}
+}
+
+func TestServer_UpdateUser_EmptyPermissions(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	username := "bob"
+	utype := client.NormalUser
+	permissions := []client.Permission{
+		client.APIPermission,
+	}
+	user, err := cli.CreateUser(client.CreateUserOptions{
+		Name:        username,
+		Password:    "hunter2",
+		Type:        utype,
+		Permissions: permissions,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if !reflect.DeepEqual(user.Permissions, permissions) {
+		t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+	}
+
+	user, err = cli.UpdateUser(user.Link, client.UpdateUserOptions{
+		Permissions: []client.Permission{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	user, err = cli.User(user.Link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if got, exp := len(user.Permissions), 0; got != exp {
+		t.Fatalf("unexpected permission count got %d exp %d", got, exp)
+	}
+}
+
+func TestServer_UpdateUser_Password_IgnorePerms(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	username := "bob"
+	utype := client.NormalUser
+	password := "hunter2"
+	permissions := []client.Permission{client.WritePointsPermission}
+	user, err := cli.CreateUser(client.CreateUserOptions{
+		Name:        username,
+		Password:    password,
+		Type:        utype,
+		Permissions: permissions,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PingAsUser(username, password); err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if !reflect.DeepEqual(user.Permissions, permissions) {
+		t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+	}
+
+	// Update password only, permissions should remain the same.
+	password = "*******"
+	user, err = cli.UpdateUser(user.Link, client.UpdateUserOptions{
+		Password: password,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PingAsUser(username, password); err != nil {
+		t.Fatal(err)
+	}
+
+	user, err = cli.User(user.Link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if !reflect.DeepEqual(user.Permissions, permissions) {
+		t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+	}
+}
+
+func TestServer_UpdateUser_Type(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	username := "bob"
+	utype := client.AdminUser
+	user, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     username,
+		Password: "hunter2",
+		Type:     utype,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if got, exp := len(user.Permissions), 0; got != exp {
+		t.Fatalf("unexpected permission count got %d exp %d", got, exp)
+	}
+
+	utype = client.NormalUser
+	user, err = cli.UpdateUser(user.Link, client.UpdateUserOptions{
+		Type: utype,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	user, err = cli.User(user.Link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if got, exp := len(user.Permissions), 0; got != exp {
+		t.Fatalf("unexpected permission count got %d exp %d", got, exp)
+	}
+}
+
+func TestServer_DeleteUser(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	username := "bob"
+	utype := client.AdminUser
+	user, err := cli.CreateUser(client.CreateUserOptions{
+		Name:     username,
+		Password: "hunter2",
+		Type:     utype,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := user.Name, username; got != exp {
+		t.Fatalf("unexpected username got %s exp %s", got, exp)
+	}
+	if got, exp := user.Link.Href, "/kapacitor/v1/users/bob"; got != exp {
+		t.Fatalf("unexpected link got %s exp %s", got, exp)
+	}
+	if got, exp := user.Type, utype; got != exp {
+		t.Fatalf("unexpected type got %v exp %v", got, exp)
+	}
+	if got, exp := len(user.Permissions), 0; got != exp {
+		t.Fatalf("unexpected permission count got %d exp %d", got, exp)
+	}
+
+	if err := cli.DeleteUser(user.Link); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := cli.User(user.Link); err == nil {
+		t.Fatal("expected error for non existent user")
+	}
+}
+func TestServer_ListUsers(t *testing.T) {
+	t.Parallel()
+	config := NewConfig()
+	config.Auth = auth.NewEnabledConfig()
+	s := OpenServer(config)
+	cli := Client(s)
+	defer s.Close()
+
+	utype := client.NormalUser
+	permissions := []client.Permission{
+		client.APIPermission,
+	}
+	count := 20
+	for i := 0; i < count; i++ {
+		username := fmt.Sprintf("bob%02d", i)
+		user, err := cli.CreateUser(client.CreateUserOptions{
+			Name:        username,
+			Password:    "hunter2",
+			Type:        utype,
+			Permissions: permissions,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got, exp := user.Name, username; got != exp {
+			t.Fatalf("unexpected username got %s exp %s", got, exp)
+		}
+		if got, exp := user.Link.Href, fmt.Sprintf("/kapacitor/v1/users/%s", username); got != exp {
+			t.Fatalf("unexpected link got %s exp %s", got, exp)
+		}
+		if got, exp := user.Type, utype; got != exp {
+			t.Fatalf("unexpected type got %v exp %v", got, exp)
+		}
+		if !reflect.DeepEqual(user.Permissions, permissions) {
+			t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+		}
+	}
+
+	// List only bob* users, aka exclude the root user
+	users, err := cli.ListUsers(&client.ListUsersOptions{
+		Pattern: "bob*",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := len(users), count; got != exp {
+		t.Fatalf("unexpected user count got %d exp %d", got, exp)
+	}
+	for i, user := range users {
+		username := fmt.Sprintf("bob%02d", i)
+		if got, exp := user.Name, username; got != exp {
+			t.Fatalf("unexpected username got %s exp %s", got, exp)
+		}
+		if got, exp := user.Link.Href, fmt.Sprintf("/kapacitor/v1/users/%s", username); got != exp {
+			t.Fatalf("unexpected link got %s exp %s", got, exp)
+		}
+		if got, exp := user.Type, utype; got != exp {
+			t.Fatalf("unexpected type got %v exp %v", got, exp)
+		}
+		if !reflect.DeepEqual(user.Permissions, permissions) {
+			t.Fatalf("unexpected permissions got %s exp %s", user.Permissions, permissions)
+		}
+	}
+
+	// List only bob1* users at +2 offset
+	users, err = cli.ListUsers(&client.ListUsersOptions{
+		Pattern: "bob1*",
+		Fields:  []string{"type"},
+		Offset:  2,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, exp := len(users), 8; got != exp {
+		t.Fatalf("unexpected user count got %d exp %d", got, exp)
+	}
+	for i, user := range users {
+		username := fmt.Sprintf("bob%02d", i+12)
+		if got, exp := user.Name, username; got != exp {
+			t.Fatalf("unexpected username got %s exp %s", got, exp)
+		}
+		if got, exp := user.Link.Href, fmt.Sprintf("/kapacitor/v1/users/%s", username); got != exp {
+			t.Fatalf("unexpected link got %s exp %s", got, exp)
+		}
+		if got, exp := user.Type, utype; got != exp {
+			t.Fatalf("unexpected type got %v exp %v", got, exp)
+		}
+	}
+}
diff --git a/services/auth/service_internal_test.go b/services/auth/service_internal_test.go
new file mode 100644
index 00000000..ee957c84
--- /dev/null
+++ b/services/auth/service_internal_test.go
@@ -0,0 +1,181 @@
+package auth
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/influxdata/kapacitor/services/auth/meta"
+)
+
+func Test_convertPMPerissions(t *testing.T) {
+	testCases := []struct {
+		scopedPerms map[string][]meta.Permission
+		exp         map[string][]Privilege
+	}{
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.KapacitorAPIPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api":        {AllPrivileges},
+				"/api/config": {NoPrivileges},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.KapacitorAPIPermission,
+					meta.KapacitorConfigAPIPermission,
+					meta.WriteDataPermission,
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":   {AllPrivileges},
+				"/api":        {AllPrivileges},
+				"/api/config": {AllPrivileges},
+				"/api/write":  {WritePrivilege},
+				"/database":   {WritePrivilege, ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.KapacitorAPIPermission,
+					meta.WriteDataPermission,
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":   {AllPrivileges},
+				"/api":        {AllPrivileges},
+				"/api/config": {NoPrivileges},
+				"/api/write":  {WritePrivilege},
+				"/database":   {WritePrivilege, ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.KapacitorConfigAPIPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":   {AllPrivileges},
+				"/api/config": {AllPrivileges},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.WriteDataPermission,
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":  {AllPrivileges},
+				"/api/write": {WritePrivilege},
+				"/database":  {WritePrivilege, ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.WriteDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":  {AllPrivileges},
+				"/api/write": {WritePrivilege},
+				"/database":  {WritePrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"": {
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping": {AllPrivileges},
+				"/database": {ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"mydb": {
+					meta.WriteDataPermission,
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":            {AllPrivileges},
+				"/api/write":           {WritePrivilege},
+				"/database/mydb_clean": {WritePrivilege, ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"mydb": {
+					meta.WriteDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":            {AllPrivileges},
+				"/api/write":           {WritePrivilege},
+				"/database/mydb_clean": {WritePrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"mydb": {
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":            {AllPrivileges},
+				"/database/mydb_clean": {ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"mydb": {
+					meta.WriteDataPermission,
+				},
+				"otherdb": {
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":               {AllPrivileges},
+				"/api/write":              {WritePrivilege},
+				"/database/mydb_clean":    {WritePrivilege},
+				"/database/otherdb_clean": {ReadPrivilege},
+			},
+		},
+		{
+			scopedPerms: map[string][]meta.Permission{
+				"mydb/slash/in/name": {
+					meta.WriteDataPermission,
+				},
+				"otherdb/moreslashes": {
+					meta.ReadDataPermission,
+				},
+			},
+			exp: map[string][]Privilege{
+				"/api/ping":                           {AllPrivileges},
+				"/api/write":                          {WritePrivilege},
+				"/database/mydb_slash_in_name_dirty":  {WritePrivilege},
+				"/database/otherdb_moreslashes_dirty": {ReadPrivilege},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		if got := convertPMPermissions(tc.scopedPerms); !reflect.DeepEqual(got, tc.exp) {
+			t.Errorf("unexpected converted privileges:\npermisions:\n%v\ngot\n%v\nexp\n%v\n", tc.scopedPerms, got, tc.exp)
+		}
+	}
+}
diff --git a/services/auth/service_test.go b/services/auth/service_test.go
new file mode 100644
index 00000000..827d7941
--- /dev/null
+++ b/services/auth/service_test.go
@@ -0,0 +1,42 @@
+package auth_test
+
+import (
+	"testing"
+
+	"golang.org/x/crypto/bcrypt"
+)
+
+var password = []byte("sorta long password")
+
+func Benchmark_Bcrypt_Cost_Min(b *testing.B) {
+	// NOTE: Min == 4
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, bcrypt.MinCost)
+	}
+}
+func Benchmark_Bcrypt_Cost_Default(b *testing.B) {
+	// NOTE: Default == 10
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, bcrypt.DefaultCost)
+	}
+}
+func Benchmark_Bcrypt_Cost_11(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, 11)
+	}
+}
+func Benchmark_Bcrypt_Cost_12(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, 12)
+	}
+}
+func Benchmark_Bcrypt_Cost_13(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, 13)
+	}
+}
+func Benchmark_Bcrypt_Cost_14(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		bcrypt.GenerateFromPassword(password, 14)
+	}
+}
diff --git a/services/kafka/kafkatest/server.go b/services/kafka/kafkatest/server.go
index 96c0e06d..6e4d34ae 100644
--- a/services/kafka/kafkatest/server.go
+++ b/services/kafka/kafkatest/server.go
@@ -27,13 +27,14 @@ type Server struct {
 	brokerMessage []byte
 	nodeID        int32
 
-	partitionMessage []byte
+	partitionCount int32
 }
 
 func NewServer() (*Server, error) {
 	s := &Server{
-		closing: make(chan struct{}),
-		nodeID:  1,
+		closing:        make(chan struct{}),
+		nodeID:         1,
+		partitionCount: 3,
 	}
 	l, err := net.Listen("tcp", ":0")
 	if err != nil {
@@ -43,7 +44,6 @@ func NewServer() (*Server, error) {
 
 	// Prepare static message bytes
 	s.prepareBrokerMsg()
-	s.preparePartitionMsg()
 
 	// start server
 	s.wg.Add(1)
@@ -64,20 +64,6 @@ func (s *Server) prepareBrokerMsg() {
 	s.brokerMessage = writeInt16(s.brokerMessage, -1)
 }
 
-func (s *Server) preparePartitionMsg() {
-	s.partitionMessage = make([]byte, 0, 2+4+4+4+4)
-	// Write error code
-	s.partitionMessage = writeInt16(s.partitionMessage, 0)
-	// Write partition ID
-	s.partitionMessage = writeInt32(s.partitionMessage, 1)
-	// Write leader ID
-	s.partitionMessage = writeInt32(s.partitionMessage, s.nodeID)
-	// Write 0 len replicas
-	s.partitionMessage = writeArrayHeader(s.partitionMessage, 0)
-	// Write 0 len Isr
-	s.partitionMessage = writeArrayHeader(s.partitionMessage, 0)
-}
-
 func (s *Server) Close() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -164,16 +150,19 @@ func (s *Server) handle(c net.Conn) error {
 
 	switch apiKey {
 	case 0: // ProduceRequest
-		topic, partition, offset := s.readProduceRequest(request)
+		topic, responses := s.readProduceRequest(request)
 
 		// Prepare success response
 		response = writeArrayHeader(response, 1)
 		response = writeStr(response, topic)
-		response = writeArrayHeader(response, 1)
-		response = writeInt32(response, partition)
-		response = writeInt16(response, 0) // Error Code
-		response = writeInt64(response, offset)
-		response = writeInt32(response, 0) // ThrottleTime
+		response = writeArrayHeader(response, int32(len(responses)))
+		for _, r := range responses {
+			response = writeInt32(response, r.partition)
+			response = writeInt16(response, 0) // Error Code
+			response = writeInt64(response, r.offset)
+			response = writeInt32(response, 0) // ThrottleTime
+		}
+
 	case 3: // Metadata
 		topics, _ := readStrList(request)
 
@@ -193,8 +182,20 @@ func (s *Server) handle(c net.Conn) error {
 			// Write is_internal
 			response = writeBool(response, false)
 
-			// Write partition
-			response = writeArray(response, [][]byte{s.partitionMessage})
+			// Write partitions
+			response = writeArrayHeader(response, s.partitionCount)
+			for i := int32(0); i < s.partitionCount; i++ {
+				// Write error code
+				response = writeInt16(response, 0)
+				// Write partition ID
+				response = writeInt32(response, i+1)
+				// Write leader ID
+				response = writeInt32(response, s.nodeID)
+				// Write 0 len replicas
+				response = writeArrayHeader(response, 0)
+				// Write 0 len Isr
+				response = writeArrayHeader(response, 0)
+			}
 		}
 	case 18:
 		// Hard code the api versions we are implementing to ensure
@@ -224,44 +225,57 @@ func (s *Server) handle(c net.Conn) error {
 	return err
 }
 
-// readProduceRequest, assume only a single message exists
-func (s *Server) readProduceRequest(request []byte) (topic string, partition int32, offset int64) {
+type produceResponse struct {
+	partition int32
+	offset    int64
+}
+
+// readProduceRequest, assume only a single message per partition exists
+func (s *Server) readProduceRequest(request []byte) (string, []produceResponse) {
+	buf := []produceResponse{}
+
 	pos := 2 + 4 + 4 // skip RequiredAcks and Timeout and array len
 
 	// Read topic name
 	topic, n := readStr(request[pos:])
 	pos += n
 
-	pos += 4 // skip array len
-
-	partition = readInt32(request[pos:])
+	// Read array len
+	arrayLen := readInt32(request[pos:])
 	pos += 4
 
-	pos += 4 // skip set size
+	for i := int32(0); i < arrayLen; i++ {
+		partition := readInt32(request[pos:])
+		pos += 4
 
-	offset = readInt64(request[pos:])
-	pos += 8
+		pos += 4 // skip set size
 
-	pos += 4 + 4 + 1 + 1 // skip size, crc, magic, attributes
+		offset := readInt64(request[pos:])
+		pos += 8
 
-	msecs := readInt64(request[pos:])
-	pos += 8
+		pos += 4 + 4 + 1 + 1 // skip size, crc, magic, attributes
 
-	key, n := readByteArray(request[pos:])
-	pos += n
+		msecs := readInt64(request[pos:])
+		pos += 8
 
-	message, n := readByteArray(request[pos:])
-	pos += n
+		key, n := readByteArray(request[pos:])
+		pos += n
 
-	s.saveMessage(Message{
-		Topic:     topic,
-		Partition: partition,
-		Offset:    offset,
-		Key:       string(key),
-		Message:   string(message),
-		Time:      time.Unix(msecs/1000, msecs%1000*1000000).UTC(),
-	})
-	return
+		message, n := readByteArray(request[pos:])
+		pos += n
+
+		s.saveMessage(Message{
+			Topic:     topic,
+			Partition: partition,
+			Offset:    offset,
+			Key:       string(key),
+			Message:   string(message),
+			Time:      time.Unix(msecs/1000, msecs%1000*1000000).UTC(),
+		})
+		buf = append(buf, produceResponse{partition, offset})
+	}
+
+	return topic, buf
 }
 
 func (s *Server) saveMessage(m Message) {
@@ -293,6 +307,9 @@ func readStr(buf []byte) (string, int) {
 }
 func readByteArray(buf []byte) ([]byte, int) {
 	n := int(int32(binary.BigEndian.Uint32(buf[:4])))
+	if n == -1 {
+		return nil, n + 4
+	}
 	return buf[4 : 4+n], n + 4
 }
 
diff --git a/services/kafka/service_test.go b/services/kafka/service_test.go
index a3d059d4..2a7c008d 100644
--- a/services/kafka/service_test.go
+++ b/services/kafka/service_test.go
@@ -26,7 +26,12 @@ func TestWriter_UpdateConfig(t *testing.T) {
 	defer diag.Close()
 
 	// Write a message to generate a writer.
-	require.NoError(t, cluster.WriteMessage(diag.NewKafkaHandler(), "testTopic", []byte{1, 2, 3, 4}, []byte{1, 2, 3, 4}))
+	require.NoError(t, cluster.WriteMessage(diag.NewKafkaHandler(),
+		kafka.WriteTarget{
+			Topic:              "testTopic",
+			PartitionById:      false,
+			PartitionAlgorithm: "",
+		}, []byte{1, 2, 3, 4}, []byte{1, 2, 3, 4}))
 
 	// Update the config in a way that requires shutting down all active writers.
 	c.UseSSL = true
diff --git a/services/zenoss/zenosstest/zenosstest.go b/services/zenoss/zenosstest/zenosstest.go
new file mode 100644
index 00000000..af195dff
--- /dev/null
+++ b/services/zenoss/zenosstest/zenosstest.go
@@ -0,0 +1,54 @@
+package zenosstest
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+
+	"github.com/influxdata/kapacitor/services/zenoss"
+)
+
+type Server struct {
+	mu       sync.Mutex
+	ts       *httptest.Server
+	URL      string
+	requests []Request
+	closed   bool
+}
+
+func NewServer() *Server {
+	s := new(Server)
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		pr := Request{
+			URL: r.URL.String(),
+		}
+		dec := json.NewDecoder(r.Body)
+		dec.Decode(&pr.Event)
+		s.mu.Lock()
+		s.requests = append(s.requests, pr)
+		s.mu.Unlock()
+	}))
+	s.ts = ts
+	s.URL = ts.URL
+	return s
+}
+
+func (s *Server) Requests() []Request {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.requests
+}
+
+func (s *Server) Close() {
+	if s.closed {
+		return
+	}
+	s.closed = true
+	s.ts.Close()
+}
+
+type Request struct {
+	URL   string
+	Event zenoss.Event
+}
diff --git a/udf/server_test.go b/udf/server_test.go
index b88b8bfc..84979bc5 100644
--- a/udf/server_test.go
+++ b/udf/server_test.go
@@ -336,7 +336,7 @@ func TestUDF_SnapshotRestore(t *testing.T) {
 		// Snapshot
 		req = <-u.Requests
 		if req == nil {
-			t.Fatal("expected snapshot message got nil")
+			t.Error("expected snapshot message got nil")
 		}
 		_, ok = req.Message.(*agent.Request_Snapshot)
 		if !ok {
@@ -351,7 +351,7 @@ func TestUDF_SnapshotRestore(t *testing.T) {
 		// Restore
 		req = <-u.Requests
 		if req == nil {
-			t.Fatal("expected restore message got nil")
+			t.Error("expected restore message got nil")
 		}
 		restore, ok := req.Message.(*agent.Request_Restore)
 		if !ok {