Add initial AMR servicenow integration (#44875)

* Add initial AMR servicenow integration

* Update servicenow AMR integration to work with RuleHandler

* Remove unused context

* Remove unused handlerTimeout

* Use watcher that allows confirming of allowed kinds

* Add tests for servicenow amr interactions

* Fix formating

integrations/access/servicenow/app.go

@@ -33,6 +33,7 @@ import (
 	"github.com/gravitational/teleport/api/accessrequest"
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/integrations/access/accessmonitoring"
 	"github.com/gravitational/teleport/integrations/access/common"
 	"github.com/gravitational/teleport/integrations/access/common/teleport"
 	"github.com/gravitational/teleport/integrations/lib"
@@ -49,8 +50,6 @@ const (
 	minServerVersion = "13.0.0"
 	// initTimeout is used to bound execution time of health check and teleport version check.
 	initTimeout = time.Second * 10
-	// handlerTimeout is used to bound the execution time of watcher event handler.
-	handlerTimeout = time.Second * 30
 	// modifyPluginDataBackoffBase is an initial (minimum) backoff value.
 	modifyPluginDataBackoffBase = time.Millisecond
 	// modifyPluginDataBackoffMax is a backoff threshold
@@ -62,11 +61,12 @@ type App struct {
 	*lib.Process
 	common.BaseApp
 
-	PluginName string
-	teleport   teleport.Client
-	serviceNow ServiceNowClient
-	mainJob    lib.ServiceJob
-	conf       Config
+	PluginName            string
+	teleport              teleport.Client
+	serviceNow            ServiceNowClient
+	mainJob               lib.ServiceJob
+	conf                  Config
+	accessMonitoringRules *accessmonitoring.RuleHandler
 }
 
 // NewServicenowApp initializes a new teleport-servicenow app and returns it.
@@ -75,6 +75,21 @@ func NewServiceNowApp(ctx context.Context, conf *Config) (*App, error) {
 		PluginName: pluginName,
 		conf:       *conf,
 	}
+	teleClient, err := conf.GetTeleportClient(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	serviceNowApp.accessMonitoringRules = accessmonitoring.NewRuleHandler(accessmonitoring.RuleHandlerConfig{
+		Client:     teleClient,
+		PluginType: string(conf.PluginType),
+		FetchRecipientCallback: func(_ context.Context, name string) (*common.Recipient, error) {
+			return &common.Recipient{
+				Name: name,
+				ID:   name,
+				Kind: common.RecipientKindSchedule,
+			}, nil
+		},
+	})
 	serviceNowApp.mainJob = lib.NewServiceJob(serviceNowApp.run)
 	return serviceNowApp, nil
 }
@@ -105,24 +120,36 @@ func (a *App) run(ctx context.Context) error {
 	if err := a.init(ctx); err != nil {
 		return trace.Wrap(err)
 	}
+	watchKinds := []types.WatchKind{
+		{Kind: types.KindAccessRequest},
+		{Kind: types.KindAccessMonitoringRule},
+	}
 
-	watcherJob, err := watcherjob.NewJob(
+	acceptedWatchKinds := make([]string, 0, len(watchKinds))
+	watcherJob, err := watcherjob.NewJobWithConfirmedWatchKinds(
 		a.teleport,
 		watcherjob.Config{
-			Watch:            types.Watch{Kinds: []types.WatchKind{{Kind: types.KindAccessRequest}}},
-			EventFuncTimeout: handlerTimeout,
+			Watch: types.Watch{Kinds: watchKinds, AllowPartialSuccess: true},
 		},
 		a.onWatcherEvent,
+		func(ws types.WatchStatus) {
+			for _, watchKind := range ws.GetKinds() {
+				acceptedWatchKinds = append(acceptedWatchKinds, watchKind.Kind)
+			}
+		},
 	)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+
 	a.SpawnCriticalJob(watcherJob)
 	ok, err := watcherJob.WaitReady(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
-
+	if err := a.accessMonitoringRules.InitAccessMonitoringRulesCache(ctx); err != nil {
+		return trace.Wrap(err)
+	}
 	a.mainJob.SetReady(ok)
 	if ok {
 		log.Info("ServiceNow plugin is ready")
@@ -187,7 +214,19 @@ func (a *App) checkTeleportVersion(ctx context.Context) (proto.PingResponse, err
 	return pong, trace.Wrap(err)
 }
 
+// onWatcherEvent is called for every cluster Event. It will filter out non-access-request events and
+// call onPendingRequest, onResolvedRequest and on DeletedRequest depending on the event.
 func (a *App) onWatcherEvent(ctx context.Context, event types.Event) error {
+	switch event.Resource.GetKind() {
+	case types.KindAccessMonitoringRule:
+		return trace.Wrap(a.accessMonitoringRules.HandleAccessMonitoringRule(ctx, event))
+	case types.KindAccessRequest:
+		return trace.Wrap(a.handleAccessRequest(ctx, event))
+	}
+	return trace.BadParameter("unexpected kind %s", event.Resource.GetKind())
+}
+
+func (a *App) handleAccessRequest(ctx context.Context, event types.Event) error {
 	if kind := event.Resource.GetKind(); kind != types.KindAccessRequest {
 		return trace.Errorf("unexpected kind %s", kind)
 	}
@@ -264,6 +303,14 @@ func (a *App) onPendingRequest(ctx context.Context, req types.AccessRequest) err
 
 	if isNew {
 		log.Infof("Creating servicenow incident")
+		recipientAssignee := a.accessMonitoringRules.RecipientsFromAccessMonitoringRules(ctx, req)
+		assignees := []string{}
+		recipientAssignee.ForEach(func(r common.Recipient) {
+			assignees = append(assignees, r.Name)
+		})
+		if len(assignees) > 0 {
+			reqData.SuggestedReviewers = assignees
+		}
 		if err = a.createIncident(ctx, reqID, reqData); err != nil {
 			// Even if we failed to create the incident we try to auto-approve
 			return trace.NewAggregate(

integrations/access/servicenow/testlib/suite.go

@@ -26,6 +26,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	accessmonitoringrulesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	v1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/integrations/access/servicenow"
 	"github.com/gravitational/teleport/integrations/lib/logger"
@@ -75,6 +77,7 @@ func (s *ServiceNowBaseSuite) SetupTest() {
 
 	var conf servicenow.Config
 	conf.Teleport = s.TeleportConfig()
+	conf.PluginType = "servicenow"
 	conf.ClientConfig.APIEndpoint = s.fakeServiceNow.URL()
 	conf.ClientConfig.CloseCode = "resolved"
 
@@ -136,6 +139,55 @@ func (s *ServiceNowSuiteOSS) TestIncidentCreation() {
 	assert.Equal(t, incident.IncidentID, pluginData.IncidentID)
 }
 
+// TestMessagePostingWithAMR validates that a message is sent to each recipient
+// specified in the monitoring rule and the plugin config is ignored. It also checks that the message
+// content is correct.
+func (s *ServiceNowSuiteOSS) TestMessagePostingWithAMR() {
+	t := s.T()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	t.Cleanup(cancel)
+
+	s.startApp()
+
+	_, err := s.ClientByName(integration.RulerUserName).
+		AccessMonitoringRulesClient().
+		CreateAccessMonitoringRule(ctx, &accessmonitoringrulesv1.AccessMonitoringRule{
+			Kind:    types.KindAccessMonitoringRule,
+			Version: types.V1,
+			Metadata: &v1.Metadata{
+				Name: "test-servicenow-amr",
+			},
+			Spec: &accessmonitoringrulesv1.AccessMonitoringRuleSpec{
+				Subjects:  []string{types.KindAccessRequest},
+				Condition: "!is_empty(access_request.spec.roles)",
+				Notification: &accessmonitoringrulesv1.Notification{
+					Name: "servicenow",
+					Recipients: []string{
+						"someReviewer", // recipient 1
+					},
+				},
+			},
+		})
+	assert.NoError(t, err)
+
+	// Test execution: we create a new access request.
+	req := s.CreateAccessRequest(ctx, integration.RequesterOSSUserName, nil)
+	pluginData := s.checkPluginData(ctx, req.GetName(), func(data servicenow.PluginData) bool {
+		return data.IncidentID != ""
+	})
+
+	// Validating a new incident was created.
+	incident, err := s.fakeServiceNow.CheckNewIncident(ctx)
+	require.NoError(t, err, "no new incidents stored")
+
+	require.Equal(t, "someReviewer", incident.AssignedTo)
+
+	assert.Equal(t, incident.IncidentID, pluginData.IncidentID)
+
+	assert.NoError(t, s.ClientByName(integration.RulerUserName).
+		AccessMonitoringRulesClient().DeleteAccessMonitoringRule(ctx, "test-servicenow-amr"))
+}
+
 // TestApproval tests that when a request is approved, its corresponding incident
 // is updated to reflect the new request state and a note is added to the incident.
 func (s *ServiceNowSuiteOSS) TestApproval() {