feat(pkger): add application functionality for the notification rules…

… resource

cmd/influx/pkg.go

@@ -790,6 +790,23 @@ func (b *cmdPkgBuilder) printPkgSummary(sum pkger.Summary) {
 		})
 	}
 
+	if rules := sum.NotificationRules; len(rules) > 0 {
+		headers := []string{"ID", "Name", "Description", "Every", "Offset", "Endpoint Name", "Endpoint ID", "Endpoint Type"}
+		tablePrintFn("NOTIFICATION RULES", headers, len(rules), func(i int) []string {
+			v := rules[i]
+			return []string{
+				v.ID.String(),
+				v.Name,
+				v.Description,
+				v.Every,
+				v.Offset,
+				v.EndpointName,
+				v.EndpointID.String(),
+				v.EndpointType,
+			}
+		})
+	}
+
 	if teles := sum.TelegrafConfigs; len(teles) > 0 {
 		headers := []string{"ID", "Name", "Description"}
 		tablePrintFn("TELEGRAF CONFIGS", headers, len(teles), func(i int) []string {

cmd/influxd/launcher/launcher.go

@@ -844,7 +844,8 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 			pkger.WithCheckSVC(authorizer.NewCheckService(b.CheckService, authedURMSVC, authedOrgSVC)),
 			pkger.WithDashboardSVC(authorizer.NewDashboardService(b.DashboardService)),
 			pkger.WithLabelSVC(authorizer.NewLabelService(b.LabelService)),
-			pkger.WithNoticationEndpointSVC(authorizer.NewNotificationEndpointService(b.NotificationEndpointService, authedURMSVC, authedOrgSVC)),
+			pkger.WithNotificationEndpointSVC(authorizer.NewNotificationEndpointService(b.NotificationEndpointService, authedURMSVC, authedOrgSVC)),
+			pkger.WithNotificationRuleSVC(authorizer.NewNotificationRuleStore(b.NotificationRuleStore, authedURMSVC, authedOrgSVC)),
 			pkger.WithSecretSVC(authorizer.NewSecretService(b.SecretService)),
 			pkger.WithTelegrafSVC(authorizer.NewTelegrafConfigService(b.TelegrafService, b.UserResourceMappingService)),
 			pkger.WithVariableSVC(authorizer.NewVariableService(b.VariableService)),

cmd/influxd/launcher/pkger_test.go

@@ -47,7 +47,7 @@ func TestLauncher_Pkger(t *testing.T) {
 				LabelService: l.LabelService(t),
 				killCount:    2, // hits error on 3rd attempt at creating a mapping
 			}),
-			pkger.WithNoticationEndpointSVC(l.NotificationEndpointService(t)),
+			pkger.WithNotificationEndpointSVC(l.NotificationEndpointService(t)),
 			pkger.WithTelegrafSVC(l.TelegrafService(t)),
 			pkger.WithVariableSVC(l.VariableService(t)),
 		)
@@ -238,6 +238,14 @@ func TestLauncher_Pkger(t *testing.T) {
 		assert.Equal(t, influxdb.TaskStatusInactive, string(endpoints[0].NotificationEndpoint.GetStatus()))
 		hasLabelAssociations(t, endpoints[0].LabelAssociations, 1, "label_1")
 
+		require.Len(t, sum1.NotificationRules, 1)
+		rule := sum1.NotificationRules[0]
+		assert.NotZero(t, rule.ID)
+		assert.Equal(t, "rule_0", rule.Name)
+		assert.Equal(t, pkger.SafeID(endpoints[0].NotificationEndpoint.GetID()), rule.EndpointID)
+		assert.Equal(t, "http_none_auth_notification_endpoint", rule.EndpointName)
+		assert.Equal(t, "http", rule.EndpointType)
+
 		teles := sum1.TelegrafConfigs
 		require.Len(t, teles, 1)
 		assert.NotZero(t, teles[0].TelegrafConfig.ID)
@@ -473,7 +481,7 @@ spec:
 				pkger.WithCheckSVC(l.CheckService()),
 				pkger.WithDashboardSVC(l.DashboardService(t)),
 				pkger.WithLabelSVC(l.LabelService(t)),
-				pkger.WithNoticationEndpointSVC(l.NotificationEndpointService(t)),
+				pkger.WithNotificationEndpointSVC(l.NotificationEndpointService(t)),
 				pkger.WithTelegrafSVC(l.TelegrafService(t)),
 				pkger.WithVariableSVC(l.VariableService(t)),
 			)

mock/notification_rule_store.go

@@ -10,45 +10,83 @@ var _ influxdb.NotificationRuleStore = &NotificationRuleStore{}
 
 // NotificationRuleStore represents a service for managing notification rule data.
 type NotificationRuleStore struct {
-	OrganizationService
-	UserResourceMappingService
-	FindNotificationRuleByIDF func(ctx context.Context, id influxdb.ID) (influxdb.NotificationRule, error)
-	FindNotificationRulesF    func(ctx context.Context, filter influxdb.NotificationRuleFilter, opt ...influxdb.FindOptions) ([]influxdb.NotificationRule, int, error)
-	CreateNotificationRuleF   func(ctx context.Context, nr influxdb.NotificationRuleCreate, userID influxdb.ID) error
-	UpdateNotificationRuleF   func(ctx context.Context, id influxdb.ID, nr influxdb.NotificationRuleCreate, userID influxdb.ID) (influxdb.NotificationRule, error)
-	PatchNotificationRuleF    func(ctx context.Context, id influxdb.ID, upd influxdb.NotificationRuleUpdate) (influxdb.NotificationRule, error)
-	DeleteNotificationRuleF   func(ctx context.Context, id influxdb.ID) error
+	*OrganizationService
+	*UserResourceMappingService
+	FindNotificationRuleByIDF     func(ctx context.Context, id influxdb.ID) (influxdb.NotificationRule, error)
+	FindNotificationRuleByIDCalls SafeCount
+	FindNotificationRulesF        func(ctx context.Context, filter influxdb.NotificationRuleFilter, opt ...influxdb.FindOptions) ([]influxdb.NotificationRule, int, error)
+	FindNotificationRulesCalls    SafeCount
+	CreateNotificationRuleF       func(ctx context.Context, nr influxdb.NotificationRuleCreate, userID influxdb.ID) error
+	CreateNotificationRuleCalls   SafeCount
+	UpdateNotificationRuleF       func(ctx context.Context, id influxdb.ID, nr influxdb.NotificationRuleCreate, userID influxdb.ID) (influxdb.NotificationRule, error)
+	UpdateNotificationRuleCalls   SafeCount
+	PatchNotificationRuleF        func(ctx context.Context, id influxdb.ID, upd influxdb.NotificationRuleUpdate) (influxdb.NotificationRule, error)
+	PatchNotificationRuleCalls    SafeCount
+	DeleteNotificationRuleF       func(ctx context.Context, id influxdb.ID) error
+	DeleteNotificationRuleCalls   SafeCount
+}
+
+// NewNotificationRuleStore creats a fake notification rules tore.
+func NewNotificationRuleStore() *NotificationRuleStore {
+	return &NotificationRuleStore{
+		OrganizationService:        NewOrganizationService(),
+		UserResourceMappingService: NewUserResourceMappingService(),
+		FindNotificationRuleByIDF: func(ctx context.Context, id influxdb.ID) (influxdb.NotificationRule, error) {
+			return nil, nil
+		},
+		FindNotificationRulesF: func(ctx context.Context, f influxdb.NotificationRuleFilter, _ ...influxdb.FindOptions) ([]influxdb.NotificationRule, int, error) {
+			return nil, 0, nil
+		},
+		CreateNotificationRuleF: func(ctx context.Context, nr influxdb.NotificationRuleCreate, userID influxdb.ID) error {
+			return nil
+		},
+		UpdateNotificationRuleF: func(ctx context.Context, id influxdb.ID, nr influxdb.NotificationRuleCreate, userID influxdb.ID) (influxdb.NotificationRule, error) {
+			return nil, nil
+		},
+		PatchNotificationRuleF: func(ctx context.Context, id influxdb.ID, upd influxdb.NotificationRuleUpdate) (influxdb.NotificationRule, error) {
+			return nil, nil
+		},
+		DeleteNotificationRuleF: func(ctx context.Context, id influxdb.ID) error {
+			return nil
+		},
+	}
 }
 
 // FindNotificationRuleByID returns a single telegraf config by ID.
 func (s *NotificationRuleStore) FindNotificationRuleByID(ctx context.Context, id influxdb.ID) (influxdb.NotificationRule, error) {
+	defer s.FindNotificationRuleByIDCalls.IncrFn()()
 	return s.FindNotificationRuleByIDF(ctx, id)
 }
 
 // FindNotificationRules returns a list of notification rules that match filter and the total count of matching notification rules.
 // Additional options provide pagination & sorting.
 func (s *NotificationRuleStore) FindNotificationRules(ctx context.Context, filter influxdb.NotificationRuleFilter, opt ...influxdb.FindOptions) ([]influxdb.NotificationRule, int, error) {
+	defer s.FindNotificationRulesCalls.IncrFn()()
 	return s.FindNotificationRulesF(ctx, filter, opt...)
 }
 
 // CreateNotificationRule creates a new notification rule and sets ID with the new identifier.
 func (s *NotificationRuleStore) CreateNotificationRule(ctx context.Context, nr influxdb.NotificationRuleCreate, userID influxdb.ID) error {
+	defer s.CreateNotificationRuleCalls.IncrFn()()
 	return s.CreateNotificationRuleF(ctx, nr, userID)
 }
 
 // UpdateNotificationRule updates a single notification rule.
 // Returns the new notification rule after update.
 func (s *NotificationRuleStore) UpdateNotificationRule(ctx context.Context, id influxdb.ID, nr influxdb.NotificationRuleCreate, userID influxdb.ID) (influxdb.NotificationRule, error) {
+	defer s.UpdateNotificationRuleCalls.IncrFn()()
 	return s.UpdateNotificationRuleF(ctx, id, nr, userID)
 }
 
 // PatchNotificationRule updates a single  notification rule with changeset.
 // Returns the new notification rule after update.
 func (s *NotificationRuleStore) PatchNotificationRule(ctx context.Context, id influxdb.ID, upd influxdb.NotificationRuleUpdate) (influxdb.NotificationRule, error) {
+	defer s.PatchNotificationRuleCalls.IncrFn()()
 	return s.PatchNotificationRuleF(ctx, id, upd)
 }
 
 // DeleteNotificationRule removes a notification rule by ID.
 func (s *NotificationRuleStore) DeleteNotificationRule(ctx context.Context, id influxdb.ID) error {
+	defer s.DeleteNotificationRuleCalls.IncrFn()()
 	return s.DeleteNotificationRuleF(ctx, id)
 }

notification/rule/rule.go

@@ -18,13 +18,11 @@ var typeToRule = map[string](func() influxdb.NotificationRule){
 	"http":      func() influxdb.NotificationRule { return &HTTP{} },
 }
 
-type rawRuleJSON struct {
-	Typ string `json:"type"`
-}
-
 // UnmarshalJSON will convert
 func UnmarshalJSON(b []byte) (influxdb.NotificationRule, error) {
-	var raw rawRuleJSON
+	var raw struct {
+		Typ string `json:"type"`
+	}
 	if err := json.Unmarshal(b, &raw); err != nil {
 		return nil, &influxdb.Error{
 			Msg: "unable to detect the notification type from json",

pkger/models.go

@@ -15,6 +15,7 @@ import (
 	"github.com/influxdata/influxdb/notification"
 	icheck "github.com/influxdata/influxdb/notification/check"
 	"github.com/influxdata/influxdb/notification/endpoint"
+	"github.com/influxdata/influxdb/notification/rule"
 )
 
 // Package kinds.
@@ -403,9 +404,9 @@ type DiffNotificationRule struct {
 	Name        string `json:"name"`
 	Description string `json:"description"`
 
-	// All these fields represent the relationship of the rule to the endpoint.
-	EndpointName string `json:"endpointName"`
+	// These 3 fields represent the relationship of the rule to the endpoint.
 	EndpointID   SafeID `json:"endpointID"`
+	EndpointName string `json:"endpointName"`
 	EndpointType string `json:"endpointType"`
 
 	Every           string              `json:"every"`
@@ -658,9 +659,15 @@ func (s *SummaryNotificationEndpoint) UnmarshalJSON(b []byte) error {
 // Summary types for NotificationRules which provide a summary of a pkg notification rule.
 type (
 	SummaryNotificationRule struct {
-		Name              string              `json:"name"`
-		Description       string              `json:"description"`
-		EndpointName      string              `json:"endpointName"`
+		ID          SafeID `json:"id"`
+		Name        string `json:"name"`
+		Description string `json:"description"`
+
+		// These 3 fields represent the relationship of the rule to the endpoint.
+		EndpointID   SafeID `json:"endpointID"`
+		EndpointName string `json:"endpointName"`
+		EndpointType string `json:"endpointType"`
+
 		Every             string              `json:"every"`
 		LabelAssociations []SummaryLabel      `json:"labelAssociations"`
 		Offset            string              `json:"offset"`
@@ -1495,6 +1502,7 @@ func (n mapperNotificationEndpoints) Len() int {
 }
 
 const (
+	fieldNotificationRuleChannel         = "channel"
 	fieldNotificationRuleCurrentLevel    = "currentLevel"
 	fieldNotificationRuleEndpointName    = "endpointName"
 	fieldNotificationRuleMessageTemplate = "messageTemplate"
@@ -1504,19 +1512,38 @@ const (
 )
 
 type notificationRule struct {
-	name         string
-	description  string
+	id    influxdb.ID
+	orgID influxdb.ID
+	name  string
+
+	channel     string
+	description string
+	every       time.Duration
+	msgTemplate string
+	offset      time.Duration
+	status      string
+	statusRules []struct{ curLvl, prevLvl string }
+	tagRules    []struct{ k, v, op string }
+
+	endpointID   influxdb.ID
 	endpointName string
-	every        time.Duration
-	msgTemplate  string
-	offset       time.Duration
-	status       string
-	statusRules  []struct{ curLvl, prevLvl string }
-	tagRules     []struct{ k, v, op string }
+	endpointType string
 
 	labels sortedLabels
 }
 
+func (r *notificationRule) Exists() bool {
+	return false
+}
+
+func (r *notificationRule) ID() influxdb.ID {
+	return r.id
+}
+
+func (r *notificationRule) Labels() []*label {
+	return r.labels
+}
+
 func (r *notificationRule) Name() string {
 	return r.name
 }
@@ -1534,8 +1561,11 @@ func (r *notificationRule) Status() influxdb.Status {
 
 func (r *notificationRule) summarize() SummaryNotificationRule {
 	return SummaryNotificationRule{
+		ID:                SafeID(r.ID()),
 		Name:              r.Name(),
+		EndpointID:        SafeID(r.endpointID),
 		EndpointName:      r.endpointName,
+		EndpointType:      r.endpointType,
 		Description:       r.description,
 		Every:             r.every.String(),
 		LabelAssociations: toSummaryLabels(r.labels...),
@@ -1547,6 +1577,55 @@ func (r *notificationRule) summarize() SummaryNotificationRule {
 	}
 }
 
+func (r *notificationRule) toInfluxRule() influxdb.NotificationRule {
+	base := rule.Base{
+		ID:          r.ID(),
+		Name:        r.Name(),
+		Description: r.description,
+		EndpointID:  r.endpointID,
+		OrgID:       r.orgID,
+		Every:       toNotificationDuration(r.every),
+		Offset:      toNotificationDuration(r.offset),
+	}
+	for _, sr := range r.statusRules {
+		var prevLvl *notification.CheckLevel
+		if lvl := notification.ParseCheckLevel(sr.prevLvl); lvl != notification.Unknown {
+			prevLvl = &lvl
+		}
+		base.StatusRules = append(base.StatusRules, notification.StatusRule{
+			CurrentLevel:  notification.ParseCheckLevel(sr.curLvl),
+			PreviousLevel: prevLvl,
+		})
+	}
+	for _, tr := range r.tagRules {
+		op, _ := influxdb.ToOperator(tr.op)
+		base.TagRules = append(base.TagRules, notification.TagRule{
+			Tag: influxdb.Tag{
+				Key:   tr.k,
+				Value: tr.v,
+			},
+			Operator: op,
+		})
+	}
+
+	switch r.endpointType {
+	case "http":
+		return &rule.HTTP{Base: base}
+	case "pagerduty":
+		return &rule.PagerDuty{
+			Base:            base,
+			MessageTemplate: r.msgTemplate,
+		}
+	case "slack":
+		return &rule.Slack{
+			Base:            base,
+			Channel:         r.channel,
+			MessageTemplate: r.msgTemplate,
+		}
+	}
+	return nil
+}
+
 func (r *notificationRule) valid() []validationErr {
 	var vErrs []validationErr
 	if r.endpointName == "" {
@@ -1601,7 +1680,7 @@ func (r *notificationRule) valid() []validationErr {
 
 	var tagErrs []validationErr
 	for i, tRule := range r.tagRules {
-		if tRule.op != "equal" {
+		if _, ok := influxdb.ToOperator(tRule.op); !ok {
 			tagErrs = append(tagErrs, validationErr{
 				Field: fieldOperator,
 				Msg:   fmt.Sprintf("must be 1 in [equal]; got=%q", tRule.op),
@@ -1659,6 +1738,16 @@ func toSummaryTagRules(tagRules []struct{ k, v, op string }) []SummaryTagRule {
 	return out
 }
 
+type mapperNotificationRules []*notificationRule
+
+func (r mapperNotificationRules) Association(i int) labelAssociater {
+	return r[i]
+}
+
+func (r mapperNotificationRules) Len() int {
+	return len(r)
+}
+
 const (
 	fieldTelegrafConfig = "config"
 )

pkger/parser.go

@@ -699,6 +699,7 @@ func (p *Pkg) graphNotificationRules() *parseErr {
 			name:         r.Name(),
 			endpointName: r.stringShort(fieldNotificationRuleEndpointName),
 			description:  r.stringShort(fieldDescription),
+			channel:      r.stringShort(fieldNotificationRuleChannel),
 			every:        r.durationShort(fieldEvery),
 			msgTemplate:  r.stringShort(fieldNotificationRuleMessageTemplate),
 			offset:       r.durationShort(fieldOffset),