Add app resource watcher/reconciler (#8228)

api/types/app.go

@@ -51,6 +51,8 @@ type Application interface {
 	GetDescription() string
 	// GetURI returns the app connection endpoint.
 	GetURI() string
+	// SetURI sets the app endpoint.
+	SetURI(string)
 	// GetPublicAddr returns the app public address.
 	GetPublicAddr() string
 	// GetInsecureSkipVerify returns the app insecure setting.
@@ -202,11 +204,16 @@ func (a *AppV3) GetDescription() string {
 	return a.Metadata.Description
 }
 
-// GetURI returns the database connection address.
+// GetURI returns the app connection address.
 func (a *AppV3) GetURI() string {
 	return a.Spec.URI
 }
 
+// SetURI sets the app connection address.
+func (a *AppV3) SetURI(uri string) {
+	a.Spec.URI = uri
+}
+
 // GetPublicAddr returns the app public address.
 func (a *AppV3) GetPublicAddr() string {
 	return a.Spec.PublicAddr
@@ -292,6 +299,14 @@ func (a Apps) Find(name string) Application {
 	return nil
 }
 
+// AsResources returns these apps as resources with labels.
+func (a Apps) AsResources() (resources ResourcesWithLabels) {
+	for _, app := range a {
+		resources = append(resources, app)
+	}
+	return resources
+}
+
 // Len returns the slice length.
 func (a Apps) Len() int { return len(a) }
 

api/types/database.go

@@ -394,6 +394,14 @@ func (d Databases) ToMap() map[string]Database {
 	return m
 }
 
+// AsResources returns these databases as resources with labels.
+func (d Databases) AsResources() (resources ResourcesWithLabels) {
+	for _, database := range d {
+		resources = append(resources, database)
+	}
+	return resources
+}
+
 // Len returns the slice length.
 func (d Databases) Len() int { return len(d) }
 

api/types/resource.go

@@ -75,6 +75,36 @@ type ResourceWithOrigin interface {
 	SetOrigin(string)
 }
 
+// ResourceWithLabels is a common interface for resources that have labels.
+type ResourceWithLabels interface {
+	// ResourceWithOrigin is the base resource interface.
+	ResourceWithOrigin
+	// GetAllLabels returns all resource's labels.
+	GetAllLabels() map[string]string
+}
+
+// ResourcesWithLabels is a list of labeled resources.
+type ResourcesWithLabels []ResourceWithLabels
+
+// Find returns resource with the specified name or nil.
+func (r ResourcesWithLabels) Find(name string) ResourceWithLabels {
+	for _, resource := range r {
+		if resource.GetName() == name {
+			return resource
+		}
+	}
+	return nil
+}
+
+// Len returns the slice length.
+func (r ResourcesWithLabels) Len() int { return len(r) }
+
+// Less compares resources by name.
+func (r ResourcesWithLabels) Less(i, j int) bool { return r[i].GetName() < r[j].GetName() }
+
+// Swap swaps two resources.
+func (r ResourcesWithLabels) Swap(i, j int) { r[i], r[j] = r[j], r[i] }
+
 // GetVersion returns resource version
 func (h *ResourceHeader) GetVersion() string {
 	return h.Version

lib/config/configuration.go

@@ -995,6 +995,14 @@ func applyAppsConfig(fc *FileConfig, cfg *service.Config) error {
 	// Enable debugging application if requested.
 	cfg.Apps.DebugApp = fc.Apps.DebugApp
 
+	// Configure resource watcher selectors if present.
+	for _, selector := range fc.Apps.Selectors {
+		cfg.Apps.Selectors = append(cfg.Apps.Selectors,
+			services.Selector{
+				MatchLabels: selector.MatchLabels,
+			})
+	}
+
 	// Loop over all apps and load app configuration.
 	for _, application := range fc.Apps.Apps {
 		// Parse the static labels of the application.
@@ -1036,7 +1044,7 @@ func applyAppsConfig(fc *FileConfig, cfg *service.Config) error {
 				Headers:  headers,
 			}
 		}
-		if err := app.Check(); err != nil {
+		if err := app.CheckAndSetDefaults(); err != nil {
 			return trace.Wrap(err)
 		}
 		cfg.Apps.Apps = append(cfg.Apps.Apps, app)
@@ -1408,7 +1416,7 @@ func Configure(clf *CommandLineFlags, cfg *service.Config) error {
 			StaticLabels:  static,
 			DynamicLabels: dynamic,
 		}
-		if err := app.Check(); err != nil {
+		if err := app.CheckAndSetDefaults(); err != nil {
 			return trace.Wrap(err)
 		}
 		cfg.Apps.Apps = append(cfg.Apps.Apps, app)

lib/config/configuration_test.go

@@ -268,7 +268,7 @@ func TestConfigReading(t *testing.T) {
 			KeyFile:  "/etc/teleport/proxy.key",
 			CertFile: "/etc/teleport/proxy.crt",
 			KeyPairs: []KeyPair{
-				KeyPair{
+				{
 					PrivateKey:  "/etc/teleport/proxy.key",
 					Certificate: "/etc/teleport/proxy.crt",
 				},
@@ -289,7 +289,7 @@ func TestConfigReading(t *testing.T) {
 				EnabledFlag: "yes",
 			},
 			Apps: []*App{
-				&App{
+				{
 					Name:          "foo",
 					URI:           "http://127.0.0.1:8080",
 					PublicAddr:    "foo.example.com",
@@ -297,6 +297,13 @@ func TestConfigReading(t *testing.T) {
 					DynamicLabels: CommandLabels,
 				},
 			},
+			Selectors: []Selector{
+				{
+					MatchLabels: map[string]apiutils.Strings{
+						"*": {"*"},
+					},
+				},
+			},
 		},
 		Databases: Databases{
 			Service: Service{
@@ -325,7 +332,7 @@ func TestConfigReading(t *testing.T) {
 				EnabledFlag:   "yes",
 			},
 			KeyPairs: []KeyPair{
-				KeyPair{
+				{
 					PrivateKey:  "/etc/teleport/proxy.key",
 					Certificate: "/etc/teleport/proxy.crt",
 				},
@@ -971,7 +978,7 @@ func makeConfigFixture() string {
 	conf.Proxy.KeyFile = "/etc/teleport/proxy.key"
 	conf.Proxy.CertFile = "/etc/teleport/proxy.crt"
 	conf.Proxy.KeyPairs = []KeyPair{
-		KeyPair{
+		{
 			PrivateKey:  "/etc/teleport/proxy.key",
 			Certificate: "/etc/teleport/proxy.crt",
 		},
@@ -993,7 +1000,7 @@ func makeConfigFixture() string {
 	// Application service.
 	conf.Apps.EnabledFlag = "yes"
 	conf.Apps.Apps = []*App{
-		&App{
+		{
 			Name:          "foo",
 			URI:           "http://127.0.0.1:8080",
 			PublicAddr:    "foo.example.com",
@@ -1001,6 +1008,11 @@ func makeConfigFixture() string {
 			DynamicLabels: CommandLabels,
 		},
 	}
+	conf.Apps.Selectors = []Selector{
+		{
+			MatchLabels: map[string]apiutils.Strings{"*": {"*"}},
+		},
+	}
 
 	// Database service.
 	conf.Databases.EnabledFlag = "yes"
@@ -1024,7 +1036,7 @@ func makeConfigFixture() string {
 	conf.Metrics.ListenAddress = "tcp://metrics"
 	conf.Metrics.CACerts = []string{"/etc/teleport/ca.crt"}
 	conf.Metrics.KeyPairs = []KeyPair{
-		KeyPair{
+		{
 			PrivateKey:  "/etc/teleport/proxy.key",
 			Certificate: "/etc/teleport/proxy.crt",
 		},
@@ -1288,6 +1300,9 @@ app_service:
       name: foo
       public_addr: "foo.example.com"
       uri: "http://127.0.0.1:8080"
+  selectors:
+  - match_labels:
+      '*': '*'
 `,
 			inComment: "config is valid",
 			outError:  false,

lib/config/fileconf.go

@@ -958,6 +958,9 @@ type Apps struct {
 
 	// Apps is a list of applications that will be run by this service.
 	Apps []*App `yaml:"apps"`
+
+	// Selectors defines resource monitor selectors.
+	Selectors []Selector `yaml:"selectors,omitempty"`
 }
 
 // App is the specific application that will be proxied by the application

lib/service/cfg.go

@@ -716,6 +716,9 @@ type AppsConfig struct {
 
 	// Apps is the list of applications that are being proxied.
 	Apps []App
+
+	// Selectors is a list of resource monitor selectors.
+	Selectors []services.Selector
 }
 
 // App is the specific application that will be proxied by the application
@@ -748,8 +751,8 @@ type App struct {
 	Rewrite *Rewrite
 }
 
-// Check validates an application.
-func (a App) Check() error {
+// CheckAndSetDefaults validates an application.
+func (a *App) CheckAndSetDefaults() error {
 	if a.Name == "" {
 		return trace.BadParameter("missing application name")
 	}
@@ -776,6 +779,11 @@ func (a App) Check() error {
 			return trace.BadParameter("application %q public_addr %q can not be an IP address, Teleport Application Access uses DNS names for routing", a.Name, a.PublicAddr)
 		}
 	}
+	// Mark the app as coming from the static configuration.
+	if a.StaticLabels == nil {
+		a.StaticLabels = make(map[string]string)
+	}
+	a.StaticLabels[types.OriginLabel] = types.OriginConfigFile
 	// Make sure there are no reserved headers in the rewrite configuration.
 	// They wouldn't be rewritten even if we allowed them here but catch it
 	// early and let the user know.

lib/service/cfg_test.go

@@ -159,7 +159,7 @@ func TestCheckApp(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {
-			err := tt.inApp.Check()
+			err := tt.inApp.CheckAndSetDefaults()
 			if tt.err != "" {
 				require.Contains(t, err.Error(), tt.err)
 			} else {

lib/service/db.go

@@ -31,6 +31,9 @@ import (
 )
 
 func (process *TeleportProcess) initDatabases() {
+	if len(process.Config.Databases.Databases) == 0 && len(process.Config.Databases.Selectors) == 0 {
+		return
+	}
 	process.registerWithAuthServer(types.RoleDatabase, DatabasesIdentityEvent)
 	process.RegisterCriticalFunc("db.init", process.initDatabaseService)
 }

lib/service/service.go

@@ -3336,7 +3336,7 @@ func (process *TeleportProcess) initApps() {
 	// If no applications are specified, exit early. This is due to the strange
 	// behavior in reading file configuration. If the user does not specify an
 	// "app_service" section, that is considered enabling "app_service".
-	if len(process.Config.Apps.Apps) == 0 && !process.Config.Apps.DebugApp {
+	if len(process.Config.Apps.Apps) == 0 && !process.Config.Apps.DebugApp && len(process.Config.Apps.Selectors) == 0 {
 		return
 	}
 
@@ -3500,6 +3500,7 @@ func (process *TeleportProcess) initApps() {
 			Hostname:     process.Config.Hostname,
 			GetRotation:  process.getRotation,
 			Apps:         applications,
+			Selectors:    process.Config.Apps.Selectors,
 			OnHeartbeat: func(err error) {
 				if err != nil {
 					process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: teleport.ComponentApp})

lib/services/reconciler.go

@@ -0,0 +1,174 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package services
+
+import (
+	"context"
+
+	"github.com/gravitational/teleport/api/types"
+
+	"github.com/gravitational/trace"
+	"github.com/sirupsen/logrus"
+)
+
+// ReconcilerConfig is the resource reconciler configuration.
+type ReconcilerConfig struct {
+	// Selectors is a list of selectors the reconciler will match resources against.
+	Selectors []Selector
+	// GetResources is used to fetch currently registered resources list.
+	GetResources func() types.ResourcesWithLabels
+	// OnCreate is called when a new resource is detected.
+	OnCreate func(context.Context, types.ResourceWithLabels) error
+	// OnUpdate is called when an existing resource is updated.
+	OnUpdate func(context.Context, types.ResourceWithLabels) error
+	// OnDelete is called when an existing resource is deleted.
+	OnDelete func(context.Context, types.ResourceWithLabels) error
+	// Log is the reconciler's logger.
+	Log logrus.FieldLogger
+}
+
+// CheckAndSetDefaults validates the reconciler configuration and sets defaults.
+func (c *ReconcilerConfig) CheckAndSetDefaults() error {
+	if c.GetResources == nil {
+		return trace.BadParameter("missing reconciler GetResources")
+	}
+	if c.OnCreate == nil {
+		return trace.BadParameter("missing reconciler OnCreate")
+	}
+	if c.OnUpdate == nil {
+		return trace.BadParameter("missing reconciler OnUpdate")
+	}
+	if c.OnDelete == nil {
+		return trace.BadParameter("missing reconciler OnDelete")
+	}
+	if c.Log == nil {
+		c.Log = logrus.WithField(trace.Component, "reconciler")
+	}
+	return nil
+}
+
+// NewReconciler creates a new reconciler with provided configuration.
+func NewReconciler(cfg ReconcilerConfig) (*Reconciler, error) {
+	if err := cfg.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &Reconciler{
+		cfg: cfg,
+		log: cfg.Log.WithField("selectors", cfg.Selectors),
+	}, nil
+}
+
+// Reconciler reconciles a list of resources returned by GetResources from its
+// config with a list of provided resources.
+//
+// It's used in combination with watchers by agents (app, database) to enable
+// dynamically registered resources.
+type Reconciler struct {
+	cfg ReconcilerConfig
+	log logrus.FieldLogger
+}
+
+// Reconcile reconciles a list of resources returned by GetResources from its
+// config with newResources and calls appropriate callbacks.
+func (r *Reconciler) Reconcile(ctx context.Context, newResources types.ResourcesWithLabels) error {
+	r.log.Debugf("Reconciling with %v resources.", len(newResources))
+	var errs []error
+
+	// Process already registered resources to see if any of them were removed.
+	for _, current := range r.cfg.GetResources() {
+		if err := r.processRegisteredResource(ctx, newResources, current); err != nil {
+			errs = append(errs, trace.Wrap(err))
+		}
+	}
+
+	// Add new resources if there are any or refresh those that were updated.
+	for _, new := range newResources {
+		if err := r.processNewResource(ctx, new); err != nil {
+			errs = append(errs, trace.Wrap(err))
+		}
+	}
+
+	return trace.NewAggregate(errs...)
+}
+
+// processRegisteredResource checks the specified registered resource against the
+// new list of resources.
+func (r *Reconciler) processRegisteredResource(ctx context.Context, newResources types.ResourcesWithLabels, registered types.ResourceWithLabels) error {
+	// Skip resources marked as static as those usually come from static config.
+	// For backwards compatibility also consider empty "origin" value.
+	if registered.Origin() == types.OriginConfigFile || registered.Origin() == "" {
+		return nil
+	}
+
+	// See if this registered resource is still present among "new" resources.
+	if new := newResources.Find(registered.GetName()); new != nil {
+		return nil
+	}
+
+	r.log.Infof("%v removed, deleting.", registered)
+	if err := r.cfg.OnDelete(ctx, registered); err != nil {
+		return trace.Wrap(err, "failed to delete %v", registered)
+	}
+
+	return nil
+}
+
+// processNewResource checks the provided new resource agsinst currently
+// registered resources.
+func (r *Reconciler) processNewResource(ctx context.Context, new types.ResourceWithLabels) error {
+	// First see if the resource is already registered and if not, whether it
+	// matches the selector labels and should be registered.
+	registered := r.cfg.GetResources().Find(new.GetName())
+	if registered == nil {
+		if MatchResourceLabels(r.cfg.Selectors, new) {
+			r.log.Infof("%v matches, creating.", new)
+			if err := r.cfg.OnCreate(ctx, new); err != nil {
+				return trace.Wrap(err, "failed to create %v", new)
+			}
+			return nil
+		}
+		r.log.Debugf("%v doesn't match, not creating.", new)
+		return nil
+	}
+
+	// Do not overwrite static resources. Consider resources with empty
+	// origin as static for backwards compatibility.
+	if registered.Origin() == types.OriginConfigFile || registered.Origin() == "" {
+		r.log.Infof("%v is part of static configuration, not creating %v.", registered, new)
+		return nil
+	}
+
+	// If the resource is already registered but was updated, see if its
+	// labels still match.
+	if new.GetResourceID() != registered.GetResourceID() {
+		if MatchResourceLabels(r.cfg.Selectors, new) {
+			r.log.Infof("%v updated, updating.", new)
+			if err := r.cfg.OnUpdate(ctx, new); err != nil {
+				return trace.Wrap(err, "failed to update %v", new)
+			}
+			return nil
+		}
+		r.log.Infof("%v updated and no longer matches, deleting.", new)
+		if err := r.cfg.OnDelete(ctx, registered); err != nil {
+			return trace.Wrap(err, "failed to delete %v", new)
+		}
+		return nil
+	}
+
+	r.log.Debugf("%v is already registered.", new)
+	return nil
+}

lib/services/reconciler_test.go

@@ -0,0 +1,238 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package services
+
+import (
+	"context"
+	"testing"
+
+	"github.com/gravitational/teleport/api/types"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestReconciler makes sure appropriate callbacks are called during reconciliation.
+func TestReconciler(t *testing.T) {
+	tests := []struct {
+		description         string
+		selectors           []Selector
+		registeredResources types.ResourcesWithLabels
+		newResources        types.ResourcesWithLabels
+		onCreateCalls       types.ResourcesWithLabels
+		onUpdateCalls       types.ResourcesWithLabels
+		onDeleteCalls       types.ResourcesWithLabels
+	}{
+		{
+			description: "new matching resource should be registered",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"*": []string{"*"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+			onCreateCalls: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+		},
+		{
+			description: "new non-matching resource should not be registered",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"env": []string{"prod"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", map[string]string{"env": "dev"}),
+			},
+		},
+		{
+			description: "resource that's no longer present should be removed",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"*": []string{"*"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+			newResources: types.ResourcesWithLabels{},
+			onDeleteCalls: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+		},
+		{
+			description: "static resource should not be overwritten",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"*": []string{"*"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{
+				makeStaticResource("res1", nil),
+			},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+		},
+		{
+			description: "resource with updated matching labels should be updated",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"*": []string{"*"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", nil),
+			},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResourceWithID("res1", map[string]string{"env": "dev"}, 1),
+			},
+			onUpdateCalls: types.ResourcesWithLabels{
+				makeDynamicResourceWithID("res1", map[string]string{"env": "dev"}, 1),
+			},
+		},
+		{
+			description: "non-matching updated resource should be removed",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"env": []string{"prod"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{
+				makeDynamicResource("res1", map[string]string{"env": "prod"}),
+			},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResourceWithID("res1", map[string]string{"env": "dev"}, 1),
+			},
+			onDeleteCalls: types.ResourcesWithLabels{
+				makeDynamicResource("res1", map[string]string{"env": "prod"}),
+			},
+		},
+		{
+			description: "complex scenario with multiple created/updated/deleted resources",
+			selectors: []Selector{{
+				MatchLabels: types.Labels{"env": []string{"prod"}},
+			}},
+			registeredResources: types.ResourcesWithLabels{
+				makeStaticResource("res0", nil),
+				makeDynamicResource("res1", map[string]string{"env": "prod"}),
+				makeDynamicResource("res2", map[string]string{"env": "prod"}),
+				makeDynamicResource("res3", map[string]string{"env": "prod"}),
+				makeDynamicResource("res4", map[string]string{"env": "prod"}),
+			},
+			newResources: types.ResourcesWithLabels{
+				makeDynamicResource("res0", map[string]string{"env": "prod"}),
+				makeDynamicResourceWithID("res2", map[string]string{"env": "prod", "a": "b"}, 1),
+				makeDynamicResource("res3", map[string]string{"env": "prod"}),
+				makeDynamicResourceWithID("res4", map[string]string{"env": "dev"}, 1),
+				makeDynamicResource("res5", map[string]string{"env": "prod"}),
+				makeDynamicResource("res6", map[string]string{"env": "dev"}),
+			},
+			onCreateCalls: types.ResourcesWithLabels{
+				makeDynamicResource("res5", map[string]string{"env": "prod"}),
+			},
+			onUpdateCalls: types.ResourcesWithLabels{
+				makeDynamicResourceWithID("res2", map[string]string{"env": "prod", "a": "b"}, 1),
+			},
+			onDeleteCalls: types.ResourcesWithLabels{
+				makeDynamicResource("res1", map[string]string{"env": "prod"}),
+				makeDynamicResource("res4", map[string]string{"env": "prod"}),
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			// Reconciler will record all callback calls in these lists.
+			var onCreateCalls, onUpdateCalls, onDeleteCalls types.ResourcesWithLabels
+
+			reconciler, err := NewReconciler(ReconcilerConfig{
+				Selectors: test.selectors,
+				GetResources: func() types.ResourcesWithLabels {
+					return test.registeredResources
+				},
+				OnCreate: func(ctx context.Context, r types.ResourceWithLabels) error {
+					onCreateCalls = append(onCreateCalls, r)
+					return nil
+				},
+				OnUpdate: func(ctx context.Context, r types.ResourceWithLabels) error {
+					onUpdateCalls = append(onUpdateCalls, r)
+					return nil
+				},
+				OnDelete: func(ctx context.Context, r types.ResourceWithLabels) error {
+					onDeleteCalls = append(onDeleteCalls, r)
+					return nil
+				},
+			})
+			require.NoError(t, err)
+
+			// Reconcile and make sure we got all expected callback calls.
+			err = reconciler.Reconcile(context.Background(), test.newResources)
+			require.NoError(t, err)
+			require.Equal(t, test.onCreateCalls, onCreateCalls)
+			require.Equal(t, test.onUpdateCalls, onUpdateCalls)
+			require.Equal(t, test.onDeleteCalls, onDeleteCalls)
+		})
+	}
+}
+
+func makeStaticResource(name string, labels map[string]string) types.ResourceWithLabels {
+	return makeResource(name, labels, map[string]string{
+		types.OriginLabel: types.OriginConfigFile,
+	}, 0)
+}
+
+func makeDynamicResource(name string, labels map[string]string) types.ResourceWithLabels {
+	return makeResource(name, labels, map[string]string{
+		types.OriginLabel: types.OriginDynamic,
+	}, 0)
+}
+
+func makeDynamicResourceWithID(name string, labels map[string]string, id int64) types.ResourceWithLabels {
+	return makeResource(name, labels, map[string]string{
+		types.OriginLabel: types.OriginDynamic,
+	}, id)
+}
+
+func makeResource(name string, labels map[string]string, additionalLabels map[string]string, id int64) types.ResourceWithLabels {
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	for k, v := range additionalLabels {
+		labels[k] = v
+	}
+	return &testResource{
+		name:   name,
+		labels: labels,
+		resID:  id,
+	}
+}
+
+type testResource struct {
+	types.ResourceWithLabels
+	name   string
+	labels map[string]string
+	resID  int64
+}
+
+func (r *testResource) GetName() string {
+	return r.name
+}
+
+func (r *testResource) Origin() string {
+	return r.labels[types.OriginLabel]
+}
+
+func (r testResource) GetResourceID() int64 {
+	return r.resID
+}
+
+func (r *testResource) GetAllLabels() map[string]string {
+	return r.labels
+}

lib/services/selector.go

@@ -38,16 +38,16 @@ func (s Selector) String() string {
 	return ""
 }
 
-// MatchDatabase returns true if any of the provided selectors matches the provided database.
-func MatchDatabase(selectors []Selector, database types.Database) bool {
+// MatchResourceLabels returns true if any of the provided selectors matches the provided database.
+func MatchResourceLabels(selectors []Selector, resource types.ResourceWithLabels) bool {
 	for _, selector := range selectors {
 		if len(selector.MatchLabels) == 0 {
 			return false
 		}
-		match, _, err := MatchLabels(selector.MatchLabels, database.GetAllLabels())
+		match, _, err := MatchLabels(selector.MatchLabels, resource.GetAllLabels())
 		if err != nil {
 			logrus.WithError(err).Errorf("Failed to match labels %v: %v.",
-				selector.MatchLabels, database)
+				selector.MatchLabels, resource)
 			return false
 		}
 		if match {

lib/services/selector_test.go

@@ -26,8 +26,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestMatchDatabase tests matching databases against a selector.
-func TestMatchDatabase(t *testing.T) {
+// TestMatchResourceLabels tests matching a resource against a selector.
+func TestMatchResourceLabels(t *testing.T) {
 	tests := []struct {
 		description    string
 		selectors      []Selector
@@ -129,7 +129,7 @@ func TestMatchDatabase(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			require.Equal(t, test.match, MatchDatabase(test.selectors, database))
+			require.Equal(t, test.match, MatchResourceLabels(test.selectors, database))
 		})
 	}
 }

lib/services/watcher.go

@@ -642,7 +642,7 @@ type DatabaseWatcherConfig struct {
 	// DatabaseGetter is responsible for fetching database resources.
 	DatabaseGetter
 	// DatabasesC receives up-to-date list of all database resources.
-	DatabasesC chan []types.Database
+	DatabasesC chan types.Databases
 }
 
 // CheckAndSetDefaults checks parameters and sets default values.
@@ -658,7 +658,7 @@ func (cfg *DatabaseWatcherConfig) CheckAndSetDefaults() error {
 		cfg.DatabaseGetter = getter
 	}
 	if cfg.DatabasesC == nil {
-		cfg.DatabasesC = make(chan []types.Database)
+		cfg.DatabasesC = make(chan types.Databases)
 	}
 	return nil
 }
@@ -684,7 +684,7 @@ type DatabaseWatcher struct {
 	*databaseCollector
 }
 
-// databaseCollector accompanies resourceWatcher when monitoring database resources..
+// databaseCollector accompanies resourceWatcher when monitoring database resources.
 type databaseCollector struct {
 	// DatabaseWatcherConfig is the watcher configuration.
 	DatabaseWatcherConfig
@@ -750,3 +750,119 @@ func databasesToSlice(databases map[string]types.Database) (slice []types.Databa
 	}
 	return slice
 }
+
+// AppWatcherConfig is an AppWatcher configuration.
+type AppWatcherConfig struct {
+	// ResourceWatcherConfig is the resource watcher configuration.
+	ResourceWatcherConfig
+	// AppGetter is responsible for fetching application resources.
+	AppGetter
+	// AppsC receives up-to-date list of all application resources.
+	AppsC chan types.Apps
+}
+
+// CheckAndSetDefaults checks parameters and sets default values.
+func (cfg *AppWatcherConfig) CheckAndSetDefaults() error {
+	if err := cfg.ResourceWatcherConfig.CheckAndSetDefaults(); err != nil {
+		return trace.Wrap(err)
+	}
+	if cfg.AppGetter == nil {
+		getter, ok := cfg.Client.(AppGetter)
+		if !ok {
+			return trace.BadParameter("missing parameter AppGetter and Client not usable as AppGetter")
+		}
+		cfg.AppGetter = getter
+	}
+	if cfg.AppsC == nil {
+		cfg.AppsC = make(chan types.Apps)
+	}
+	return nil
+}
+
+// NewAppWatcher returns a new instance of AppWatcher.
+func NewAppWatcher(ctx context.Context, cfg AppWatcherConfig) (*AppWatcher, error) {
+	if err := cfg.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	collector := &appCollector{
+		AppWatcherConfig: cfg,
+	}
+	watcher, err := newResourceWatcher(ctx, collector, cfg.ResourceWatcherConfig)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &AppWatcher{watcher, collector}, nil
+}
+
+// AppWatcher is built on top of resourceWatcher to monitor application resources.
+type AppWatcher struct {
+	*resourceWatcher
+	*appCollector
+}
+
+// appCollector accompanies resourceWatcher when monitoring application resources.
+type appCollector struct {
+	// AppWatcherConfig is the watcher configuration.
+	AppWatcherConfig
+	// current holds a map of the currently known application resources.
+	current map[string]types.Application
+	// lock protects the "current" map.
+	lock sync.RWMutex
+}
+
+// resourceKind specifies the resource kind to watch.
+func (p *appCollector) resourceKind() string {
+	return types.KindApp
+}
+
+// getResourcesAndUpdateCurrent refreshes the list of current resources.
+func (p *appCollector) getResourcesAndUpdateCurrent(ctx context.Context) error {
+	apps, err := p.AppGetter.GetApps(ctx)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	newCurrent := make(map[string]types.Application, len(apps))
+	for _, app := range apps {
+		newCurrent[app.GetName()] = app
+	}
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.current = newCurrent
+	p.AppsC <- apps
+	return nil
+}
+
+// processEventAndUpdateCurrent is called when a watcher event is received.
+func (p *appCollector) processEventAndUpdateCurrent(ctx context.Context, event types.Event) {
+	if event.Resource == nil || event.Resource.GetKind() != types.KindApp {
+		p.Log.Warnf("Unexpected event: %v.", event)
+		return
+	}
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	switch event.Type {
+	case types.OpDelete:
+		delete(p.current, event.Resource.GetName())
+		p.AppsC <- appsToSlice(p.current)
+	case types.OpPut:
+		app, ok := event.Resource.(types.Application)
+		if !ok {
+			p.Log.Warnf("Unexpected resource type %T.", event.Resource)
+			return
+		}
+		p.current[app.GetName()] = app
+		p.AppsC <- appsToSlice(p.current)
+	default:
+		p.Log.Warnf("Unsupported event type %s.", event.Type)
+		return
+	}
+}
+
+func (*appCollector) notifyStale() {}
+
+func appsToSlice(apps map[string]types.Application) (slice []types.Application) {
+	for _, app := range apps {
+		slice = append(slice, app)
+	}
+	return slice
+}

lib/services/watcher_test.go

@@ -480,7 +480,7 @@ func TestDatabaseWatcher(t *testing.T) {
 				Events:    local.NewEventsService(bk, nil),
 			},
 		},
-		DatabasesC: make(chan []types.Database, 10),
+		DatabasesC: make(chan types.Databases, 10),
 	})
 	require.NoError(t, err)
 	t.Cleanup(w.Close)
@@ -549,3 +549,99 @@ func newDatabase(t *testing.T, name string) types.Database {
 	require.NoError(t, err)
 	return database
 }
+
+// TestAppWatcher tests that application resource watcher properly receives
+// and dispatches updates.
+func TestAppWatcher(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	bk, err := lite.NewWithConfig(ctx, lite.Config{
+		Path:             t.TempDir(),
+		PollStreamPeriod: 200 * time.Millisecond,
+	})
+	require.NoError(t, err)
+
+	type client struct {
+		services.Apps
+		types.Events
+	}
+
+	appService := local.NewAppService(bk)
+	w, err := services.NewAppWatcher(ctx, services.AppWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component:   "test",
+			RetryPeriod: 200 * time.Millisecond,
+			Client: &client{
+				Apps:   appService,
+				Events: local.NewEventsService(bk, nil),
+			},
+		},
+		AppsC: make(chan types.Apps, 10),
+	})
+	require.NoError(t, err)
+	t.Cleanup(w.Close)
+
+	// Initially there are no apps so watcher should send an empty list.
+	select {
+	case changeset := <-w.AppsC:
+		require.Len(t, changeset, 0)
+	case <-w.Done():
+		t.Fatal("Watcher has unexpectedly exited.")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for the first event.")
+	}
+
+	// Add an app.
+	app1 := newApp(t, "app1")
+	require.NoError(t, appService.CreateApp(ctx, app1))
+
+	// The first event is always the current list of apps.
+	select {
+	case changeset := <-w.AppsC:
+		require.Len(t, changeset, 1)
+		require.Empty(t, resourceDiff(changeset[0], app1))
+	case <-w.Done():
+		t.Fatal("Watcher has unexpectedly exited.")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for the first event.")
+	}
+
+	// Add a second app.
+	app2 := newApp(t, "app2")
+	require.NoError(t, appService.CreateApp(ctx, app2))
+
+	// Watcher should detect the app list change.
+	select {
+	case changeset := <-w.AppsC:
+		require.Len(t, changeset, 2)
+	case <-w.Done():
+		t.Fatal("Watcher has unexpectedly exited.")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for the update event.")
+	}
+
+	// Delete the first app.
+	require.NoError(t, appService.DeleteApp(ctx, app1.GetName()))
+
+	// Watcher should detect the database list change.
+	select {
+	case changeset := <-w.AppsC:
+		require.Len(t, changeset, 1)
+		require.Empty(t, resourceDiff(changeset[0], app2))
+	case <-w.Done():
+		t.Fatal("Watcher has unexpectedly exited.")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for the update event.")
+	}
+}
+
+func newApp(t *testing.T, name string) types.Application {
+	app, err := types.NewAppV3(types.Metadata{
+		Name: name,
+	}, types.AppSpecV3{
+		URI: "localhost",
+	})
+	require.NoError(t, err)
+	return app
+}

lib/srv/app/server.go

@@ -89,6 +89,12 @@ type Config struct {
 
 	// Cloud provides cloud provider access related functionality.
 	Cloud Cloud
+
+	// Selectors is a list of resource monitor selectors.
+	Selectors []services.Selector
+
+	// OnReconcile is called after each database resource reconciliation.
+	OnReconcile func(types.Apps)
 }
 
 // CheckAndSetDefaults makes sure the configuration has the minimum required
@@ -159,6 +165,11 @@ type Server struct {
 	proxyPort string
 
 	cache *sessionCache
+
+	// watcher monitors changes to application resources.
+	watcher *services.AppWatcher
+	// reconciler reconciles proxied apps with application resources.
+	reconciler *services.Reconciler
 }
 
 // New returns a new application server.
@@ -203,6 +214,12 @@ func New(ctx context.Context, c *Config) (*Server, error) {
 	// Figure out the port the proxy is running on.
 	s.proxyPort = s.getProxyPort()
 
+	// Reconciler will be reconciling application resources with proxied apps.
+	s.reconciler, err = s.getReconciler()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	return s, nil
 }
 
@@ -378,6 +395,22 @@ func (s *Server) registerApp(ctx context.Context, app types.Application) error {
 	return nil
 }
 
+// updateApp updates application that is already registered.
+func (s *Server) updateApp(ctx context.Context, app types.Application) error {
+	// Stop heartbeat and dynamic labels before starting new ones.
+	if err := s.stopApp(ctx, app.GetName()); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := s.registerApp(ctx, app); err != nil {
+		// If we failed to re-register, don't keep proxying the old app.
+		if errUnregister := s.unregisterApp(ctx, app.GetName()); errUnregister != nil {
+			return trace.NewAggregate(err, errUnregister)
+		}
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
 // unregisterApp stops proxying the app.
 func (s *Server) unregisterApp(ctx context.Context, name string) error {
 	if err := s.stopApp(ctx, name); err != nil {
@@ -407,6 +440,22 @@ func (s *Server) Start(ctx context.Context) error {
 			return trace.Wrap(err)
 		}
 	}
+
+	// Register all matching apps from resources.
+	apps, err := s.c.AccessPoint.GetApps(ctx)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	if err := s.reconciler.Reconcile(ctx, types.Apps(apps).AsResources()); err != nil {
+		return trace.Wrap(err)
+	}
+
+	// Initialize watcher that will be dynamically (un-)registering
+	// proxied apps based on the application resources.
+	if s.watcher, err = s.startWatcher(ctx); err != nil {
+		return trace.Wrap(err)
+	}
+
 	return nil
 }
 
@@ -429,6 +478,11 @@ func (s *Server) Close() error {
 	// Signal to any blocking go routine that it should exit.
 	s.closeFunc()
 
+	// Stop the database resource watcher.
+	if s.watcher != nil {
+		s.watcher.Close()
+	}
+
 	return trace.NewAggregate(errs...)
 }
 

lib/srv/app/server_test.go

@@ -34,20 +34,20 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/gravitational/teleport"
 	"github.com/gravitational/teleport/api/constants"
-	apidefaults "github.com/gravitational/teleport/api/defaults"
+	"github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/auth"
 	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/utils"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/jonboulle/clockwork"
 	"github.com/pborman/uuid"
-	"gopkg.in/check.v1"
+	"github.com/stretchr/testify/require"
 )
 
 func TestMain(m *testing.M) {
@@ -77,13 +77,42 @@ type Suite struct {
 	role types.Role
 }
 
-var _ = check.Suite(&Suite{})
+func (s *Suite) TearDown(t *testing.T) {
+	err := s.appServer.Close()
+	require.NoError(t, err)
 
-func TestApp(t *testing.T) { check.TestingT(t) }
+	err = s.authClient.Close()
+	require.NoError(t, err)
+
+	s.testhttp.Close()
+
+	err = s.tlsServer.Auth().DeleteAllAppServers(s.closeContext, defaults.Namespace)
+	require.NoError(t, err)
+
+	s.closeFunc()
+
+	err = s.tlsServer.Close()
+	require.NoError(t, err)
+}
+
+type suiteConfig struct {
+	// Selectors are resource watcher selectors.
+	Selectors []services.Selector
+	// OnReconcile sets app resource reconciliation callback.
+	OnReconcile func(types.Apps)
+	// Apps are the apps to configure.
+	Apps types.Apps
+}
+
+func SetUpSuite(t *testing.T) *Suite {
+	return SetUpSuiteWithConfig(t, suiteConfig{})
+}
+
+func SetUpSuiteWithConfig(t *testing.T, config suiteConfig) *Suite {
+	s := &Suite{}
 
-func (s *Suite) SetUpSuite(c *check.C) {
 	s.clock = clockwork.NewFakeClock()
-	s.dataDir = c.MkDir()
+	s.dataDir = t.TempDir()
 	s.hostUUID = uuid.New()
 
 	var err error
@@ -93,50 +122,28 @@ func (s *Suite) SetUpSuite(c *check.C) {
 		Dir:         s.dataDir,
 		Clock:       s.clock,
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.tlsServer, err = s.authServer.NewTestTLSServer()
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Create user and role.
 	s.user, s.role, err = auth.CreateUserAndRole(s.tlsServer.Auth(), "foo", []string{"foo-login"})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Grant the user's role access to the application label "bar: baz".
 	s.role.SetAppLabels(services.Allow, types.Labels{"bar": []string{"baz"}})
 	s.role.SetAWSRoleARNs(services.Allow, []string{"readonly"})
 	err = s.tlsServer.Auth().UpsertRole(context.Background(), s.role)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	rootCA, err := s.tlsServer.Auth().GetCertAuthority(types.CertAuthID{
 		Type:       types.HostCA,
 		DomainName: "root.example.com",
 	}, false)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.hostCertPool, err = services.CertPool(rootCA)
-	c.Assert(err, check.IsNil)
-}
+	require.NoError(t, err)
 
-func (s *Suite) TearDownSuite(c *check.C) {
-	err := s.tlsServer.Close()
-	c.Assert(err, check.IsNil)
-}
-
-var (
-	staticLabels = map[string]string{
-		"bar": "baz",
-	}
-	dynamicLabelName    = "qux"
-	dynamicLabelPeriod  = types.NewDuration(time.Second)
-	dynamicLabelCommand = []string{"expr", "1", "+", "3"}
-	dynamicLabels       = map[string]types.CommandLabel{
-		dynamicLabelName: &types.CommandLabelV2{
-			Period:  dynamicLabelPeriod,
-			Command: dynamicLabelCommand,
-		},
-	}
-)
-
-func (s *Suite) SetUpTest(c *check.C) {
 	s.closeContext, s.closeFunc = context.WithCancel(context.Background())
 
 	// Create a in-memory HTTP server that will respond with a UUID. This value
@@ -151,7 +158,7 @@ func (s *Suite) SetUpTest(c *check.C) {
 
 	// Extract the hostport that the in-memory HTTP server is running on.
 	u, err := url.Parse(s.testhttp.URL)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.hostport = u.Host
 
 	// Create apps that will be used for each test.
@@ -163,7 +170,7 @@ func (s *Suite) SetUpTest(c *check.C) {
 		PublicAddr:    "foo.example.com",
 		DynamicLabels: types.LabelsToV2(dynamicLabels),
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	appAWS, err := types.NewAppV3(types.Metadata{
 		Name:   "awsconsole",
 		Labels: staticLabels,
@@ -171,21 +178,21 @@ func (s *Suite) SetUpTest(c *check.C) {
 		URI:        constants.AWSConsoleURL,
 		PublicAddr: "aws.example.com",
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Create a client with a machine role of RoleApp.
 	s.authClient, err = s.tlsServer.NewClient(auth.TestServerID(types.RoleApp, s.hostUUID))
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	serverIdentity, err := auth.NewServerIdentity(s.authServer.AuthServer, s.hostUUID, types.RoleApp)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	tlsConfig, err := serverIdentity.TLSConfig(nil)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	tlsConfig.Time = s.clock.Now
 
 	// Generate certificate for user.
 	privateKey, publicKey, err := s.tlsServer.Auth().GenerateKeyPair("")
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	certificate, err := s.tlsServer.Auth().GenerateUserAppTestCert(auth.AppTestCertRequest{
 		PublicKey:   publicKey,
 		Username:    s.user.GetName(),
@@ -193,13 +200,13 @@ func (s *Suite) SetUpTest(c *check.C) {
 		PublicAddr:  "foo.example.com",
 		ClusterName: "root.example.com",
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.clientCertificate, err = tls.X509KeyPair(certificate, privateKey)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Generate certificate for AWS console application.
 	privateKey, publicKey, err = s.tlsServer.Auth().GenerateKeyPair("")
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	certificate, err = s.tlsServer.Auth().GenerateUserAppTestCert(auth.AppTestCertRequest{
 		PublicKey:   publicKey,
 		Username:    s.user.GetName(),
@@ -208,16 +215,16 @@ func (s *Suite) SetUpTest(c *check.C) {
 		ClusterName: "root.example.com",
 		AWSRoleARN:  "readonly",
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.awsConsoleCertificate, err = tls.X509KeyPair(certificate, privateKey)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Make sure the upload directory is created.
 	err = os.MkdirAll(filepath.Join(
 		s.dataDir, teleport.LogsDir, teleport.ComponentUpload,
-		events.StreamingLogsDir, apidefaults.Namespace,
+		events.StreamingLogsDir, defaults.Namespace,
 	), 0755)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	lockWatcher, err := services.NewLockWatcher(s.closeContext, services.LockWatcherConfig{
 		ResourceWatcherConfig: services.ResourceWatcherConfig{
@@ -225,9 +232,14 @@ func (s *Suite) SetUpTest(c *check.C) {
 			Client:    s.authClient,
 		},
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	authorizer, err := auth.NewAuthorizer("cluster-name", s.authClient, lockWatcher)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
+
+	apps := types.Apps{appFoo, appAWS}
+	if len(config.Apps) > 0 {
+		apps = config.Apps
+	}
 
 	s.appServer, err = New(s.closeContext, &Config{
 		Clock:        s.clock,
@@ -240,39 +252,30 @@ func (s *Suite) SetUpTest(c *check.C) {
 		Hostname:     "test",
 		Authorizer:   authorizer,
 		GetRotation:  testRotationGetter,
-		Apps:         types.Apps{appFoo, appAWS},
+		Apps:         apps,
 		OnHeartbeat:  func(err error) {},
 		Cloud:        &testCloud{},
+		Selectors:    config.Selectors,
+		OnReconcile:  config.OnReconcile,
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	err = s.appServer.Start(s.closeContext)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	err = s.appServer.ForceHeartbeat()
-	c.Assert(err, check.IsNil)
-}
+	require.NoError(t, err)
 
-func (s *Suite) TearDownTest(c *check.C) {
-	err := s.appServer.Close()
-	c.Assert(err, check.IsNil)
-
-	err = s.authClient.Close()
-	c.Assert(err, check.IsNil)
-
-	s.testhttp.Close()
-
-	err = s.tlsServer.Auth().DeleteAllAppServers(s.closeContext, apidefaults.Namespace)
-	c.Assert(err, check.IsNil)
-
-	s.closeFunc()
+	return s
 }
 
 // TestStart makes sure that after the server has started, a correct services.App
 // has been created.
-func (s *Suite) TestStart(c *check.C) {
+func TestStart(t *testing.T) {
+	s := SetUpSuite(t)
+
 	// Fetch the services.App that the service heartbeat.
-	servers, err := s.authServer.AuthServer.GetApplicationServers(s.closeContext, apidefaults.Namespace)
-	c.Assert(err, check.IsNil)
+	servers, err := s.authServer.AuthServer.GetApplicationServers(s.closeContext, defaults.Namespace)
+	require.NoError(t, err)
 
 	// Check that the services.Server sent via heartbeat is correct. For example,
 	// check that the dynamic labels have been evaluated.
@@ -290,9 +293,9 @@ func (s *Suite) TestStart(c *check.C) {
 			},
 		},
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	serverFoo, err := types.NewAppServerV3FromApp(appFoo, "test", s.hostUUID)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	appAWS, err := types.NewAppV3(types.Metadata{
 		Name:   "awsconsole",
 		Labels: staticLabels,
@@ -300,23 +303,25 @@ func (s *Suite) TestStart(c *check.C) {
 		URI:        constants.AWSConsoleURL,
 		PublicAddr: "aws.example.com",
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	serverAWS, err := types.NewAppServerV3FromApp(appAWS, "test", s.hostUUID)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	sort.Sort(types.AppServers(servers))
-	c.Assert(cmp.Diff([]types.AppServer{serverAWS, serverFoo}, servers,
-		cmpopts.IgnoreFields(types.Metadata{}, "ID", "Expires")), check.Equals, "")
+	require.Empty(t, cmp.Diff([]types.AppServer{serverAWS, serverFoo}, servers,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID", "Expires")))
 
 	// Check the expiry time is correct.
 	for _, server := range servers {
-		c.Assert(s.clock.Now().Before(server.Expiry()), check.Equals, true)
-		c.Assert(s.clock.Now().Add(2*apidefaults.ServerAnnounceTTL).After(server.Expiry()), check.Equals, true)
+		require.True(t, s.clock.Now().Before(server.Expiry()))
+		require.True(t, s.clock.Now().Add(2*defaults.ServerAnnounceTTL).After(server.Expiry()))
 	}
 }
 
 // TestWaitStop makes sure the server will block and unlock.
-func (s *Suite) TestWaitStop(c *check.C) {
+func TestWaitStop(t *testing.T) {
+	s := SetUpSuite(t)
+
 	// Make sure that wait will block while the server is running.
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
@@ -325,80 +330,88 @@ func (s *Suite) TestWaitStop(c *check.C) {
 	}()
 	select {
 	case <-ctx.Done():
-		c.Fatalf("Wait failed to block.")
+		t.Fatal("Wait failed to block.")
 	case <-time.After(250 * time.Millisecond):
 	}
 
 	// Close should unblock wait.
 	err := s.appServer.Close()
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	err = s.appServer.Wait()
-	c.Assert(err, check.Equals, context.Canceled)
+	require.Equal(t, err, context.Canceled)
 }
 
 // TestHandleConnection verifies that requests with valid certificates are forwarded.
-func (s *Suite) TestHandleConnection(c *check.C) {
-	s.checkHTTPResponse(c, s.clientCertificate, func(resp *http.Response) {
-		c.Assert(resp.StatusCode, check.Equals, http.StatusOK)
+func TestHandleConnection(t *testing.T) {
+	s := SetUpSuite(t)
+	s.checkHTTPResponse(t, s.clientCertificate, func(resp *http.Response) {
+		require.Equal(t, resp.StatusCode, http.StatusOK)
 		buf, err := ioutil.ReadAll(resp.Body)
-		c.Assert(err, check.IsNil)
-		c.Assert(strings.TrimSpace(string(buf)), check.Equals, s.message)
+		require.NoError(t, err)
+		require.Equal(t, strings.TrimSpace(string(buf)), s.message)
 	})
 }
 
 // TestAuthorize verifies that only authorized requests are handled.
-func (s *Suite) TestAuthorize(c *check.C) {
+func TestAuthorize(t *testing.T) {
+	// TODO(r0mant): Implement this.
 }
 
 // TestAuthorizeWithLocks verifies that requests are forbidden when there is
 // a matching lock in force.
-func (s *Suite) TestAuthorizeWithLocks(c *check.C) {
+func TestAuthorizeWithLocks(t *testing.T) {
+	s := SetUpSuite(t)
 	// Create a lock targeting the user.
 	lock, err := types.NewLock("test-lock", types.LockSpecV2{
 		Target: types.LockTarget{User: s.user.GetName()},
 	})
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 	s.tlsServer.Auth().UpsertLock(s.closeContext, lock)
 	defer func() {
 		s.tlsServer.Auth().DeleteLock(s.closeContext, lock.GetName())
 	}()
 
-	s.checkHTTPResponse(c, s.clientCertificate, func(resp *http.Response) {
-		c.Assert(resp.StatusCode, check.Equals, http.StatusForbidden)
+	s.checkHTTPResponse(t, s.clientCertificate, func(resp *http.Response) {
+		require.Equal(t, resp.StatusCode, http.StatusForbidden)
 		buf, err := ioutil.ReadAll(resp.Body)
-		c.Assert(err, check.IsNil)
-		c.Assert(strings.TrimSpace(string(buf)), check.Equals, "Forbidden")
+		require.NoError(t, err)
+		require.Equal(t, strings.TrimSpace(string(buf)), "Forbidden")
 	})
 }
 
 // TestGetConfigForClient verifies that only the CAs of the requested cluster are returned.
-func (s *Suite) TestGetConfigForClient(c *check.C) {
+func TestGetConfigForClient(t *testing.T) {
+	// TODO(r0mant): Implement this.
 }
 
 // TestRewriteRequest verifies that requests are rewritten to include JWT headers.
-func (s *Suite) TestRewriteRequest(c *check.C) {
+func TestRewriteRequest(t *testing.T) {
+	// TODO(r0mant): Implement this.
 }
 
 // TestRewriteResponse verifies that responses are rewritten if rewrite rules are specified.
-func (s *Suite) TestRewriteResponse(c *check.C) {
+func TestRewriteResponse(t *testing.T) {
+	// TODO(r0mant): Implement this.
 }
 
 // TestSessionClose makes sure sessions are closed after the given session time period.
-func (s *Suite) TestSessionClose(c *check.C) {
+func TestSessionClose(t *testing.T) {
+	// TODO(r0mant): Implement this.
 }
 
 // TestAWSConsoleRedirect verifies AWS management console access.
-func (s *Suite) TestAWSConsoleRedirect(c *check.C) {
-	s.checkHTTPResponse(c, s.awsConsoleCertificate, func(resp *http.Response) {
-		c.Assert(resp.StatusCode, check.Equals, http.StatusFound)
+func TestAWSConsoleRedirect(t *testing.T) {
+	s := SetUpSuite(t)
+	s.checkHTTPResponse(t, s.awsConsoleCertificate, func(resp *http.Response) {
+		require.Equal(t, resp.StatusCode, http.StatusFound)
 		location, err := resp.Location()
-		c.Assert(err, check.IsNil)
-		c.Assert(location.String(), check.Equals, "https://signin.aws.amazon.com")
+		require.NoError(t, err)
+		require.Equal(t, location.String(), "https://signin.aws.amazon.com")
 	})
 }
 
 // checkHTTPResponse checks expected HTTP response.
-func (s *Suite) checkHTTPResponse(c *check.C, clientCert tls.Certificate, checkResp func(*http.Response)) {
+func (s *Suite) checkHTTPResponse(t *testing.T, clientCert tls.Certificate, checkResp func(*http.Response)) {
 	pr, pw := net.Pipe()
 	defer pw.Close()
 	defer pr.Close()
@@ -437,16 +450,16 @@ func (s *Suite) checkHTTPResponse(c *check.C, clientCert tls.Certificate, checkR
 
 	// Issue request.
 	resp, err := httpClient.Get("https://" + constants.APIDomain)
-	c.Assert(err, check.IsNil)
+	require.NoError(t, err)
 
 	// Check response.
 	checkResp(resp)
-	c.Assert(resp.Body.Close(), check.IsNil)
+	require.NoError(t, resp.Body.Close())
 
 	// Context will close because of the net.Pipe, expect a context canceled
 	// error here.
 	err = s.appServer.Close()
-	c.Assert(err, check.NotNil)
+	require.NotNil(t, err)
 
 	// Wait for the application server to actually stop serving before
 	// closing the test. This will make sure the server removes the listeners
@@ -464,3 +477,18 @@ func (c *testCloud) GetAWSSigninURL(_ AWSSigninRequest) (*AWSSigninResponse, err
 		SigninURL: "https://signin.aws.amazon.com",
 	}, nil
 }
+
+var (
+	staticLabels = map[string]string{
+		"bar": "baz",
+	}
+	dynamicLabelName    = "qux"
+	dynamicLabelPeriod  = types.NewDuration(time.Second)
+	dynamicLabelCommand = []string{"expr", "1", "+", "3"}
+	dynamicLabels       = map[string]types.CommandLabel{
+		dynamicLabelName: &types.CommandLabelV2{
+			Period:  dynamicLabelPeriod,
+			Command: dynamicLabelCommand,
+		},
+	}
+)

lib/srv/app/watcher.go

@@ -0,0 +1,99 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package app
+
+import (
+	"context"
+
+	"github.com/gravitational/teleport"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/services"
+
+	"github.com/gravitational/trace"
+)
+
+// startWatcher starts watching changes to application resources and registers /
+// unregisters the proxied applications accordingly.
+func (s *Server) startWatcher(ctx context.Context) (*services.AppWatcher, error) {
+	if len(s.c.Selectors) == 0 {
+		s.log.Debug("Not initializing application resource watcher.")
+		return nil, nil
+	}
+	s.log.Debug("Initializing application resource watcher.")
+	watcher, err := services.NewAppWatcher(ctx, services.AppWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component: teleport.ComponentApp,
+			Log:       s.log,
+			Client:    s.c.AccessPoint,
+		},
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	go func() {
+		defer watcher.Close()
+		for {
+			select {
+			case apps := <-watcher.AppsC:
+				if err := s.reconciler.Reconcile(ctx, apps.AsResources()); err != nil {
+					s.log.WithError(err).Errorf("Failed to reconcile %v.", apps)
+				} else if s.c.OnReconcile != nil {
+					s.c.OnReconcile(s.getApps())
+				}
+			case <-ctx.Done():
+				s.log.Debug("Application resource watcher done.")
+				return
+			}
+		}
+	}()
+	return watcher, nil
+}
+
+func (s *Server) getReconciler() (*services.Reconciler, error) {
+	return services.NewReconciler(services.ReconcilerConfig{
+		Selectors:    s.c.Selectors,
+		GetResources: s.getResources,
+		OnCreate:     s.onCreate,
+		OnUpdate:     s.onUpdate,
+		OnDelete:     s.onDelete,
+		Log:          s.log,
+	})
+}
+
+func (s *Server) getResources() (resources types.ResourcesWithLabels) {
+	return s.getApps().AsResources()
+}
+
+func (s *Server) onCreate(ctx context.Context, resource types.ResourceWithLabels) error {
+	app, ok := resource.(types.Application)
+	if !ok {
+		return trace.BadParameter("expected types.Application, got %T", resource)
+	}
+	return s.registerApp(ctx, app)
+}
+
+func (s *Server) onUpdate(ctx context.Context, resource types.ResourceWithLabels) error {
+	app, ok := resource.(types.Application)
+	if !ok {
+		return trace.BadParameter("expected types.Application, got %T", resource)
+	}
+	return s.updateApp(ctx, app)
+}
+
+func (s *Server) onDelete(ctx context.Context, resource types.ResourceWithLabels) error {
+	return s.unregisterApp(ctx, resource.GetName())
+}

lib/srv/app/watcher_test.go

@@ -0,0 +1,210 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package app
+
+import (
+	"context"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/services"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/stretchr/testify/require"
+)
+
+// TestWatcher verifies that app agent properly detects and applies
+// changes to application resources.
+func TestWatcher(t *testing.T) {
+	ctx := context.Background()
+
+	// Make a static configuration app.
+	app0, err := makeStaticApp("app0", nil)
+	require.NoError(t, err)
+
+	// This channel will receive new set of apps the server proxies
+	// after each reconciliation.
+	reconcileCh := make(chan types.Apps)
+
+	// Setup app server that proxies one static app and
+	// watches for apps with label group=a.
+	s := SetUpSuiteWithConfig(t, suiteConfig{
+		Apps: types.Apps{app0},
+		Selectors: []services.Selector{
+			{MatchLabels: types.Labels{
+				"group": []string{"a"},
+			}},
+		},
+		OnReconcile: func(a types.Apps) {
+			reconcileCh <- a
+		},
+	})
+
+	// Only app0 should be registered initially.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Create app with label group=a.
+	app1, err := makeDynamicApp("app1", map[string]string{"group": "a"})
+	require.NoError(t, err)
+	err = s.authServer.AuthServer.CreateApp(ctx, app1)
+	require.NoError(t, err)
+
+	// It should be registered.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app1}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Try to update app0 which is registered statically.
+	app0Updated, err := makeDynamicApp("app0", map[string]string{"group": "a", types.OriginLabel: types.OriginDynamic})
+	require.NoError(t, err)
+	err = s.authServer.AuthServer.CreateApp(ctx, app0Updated)
+	require.NoError(t, err)
+
+	// It should not be registered, old app0 should remain.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app1}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Create app with label group=b.
+	app2, err := makeDynamicApp("app2", map[string]string{"group": "b"})
+	require.NoError(t, err)
+	err = s.authServer.AuthServer.CreateApp(ctx, app2)
+	require.NoError(t, err)
+
+	// It shouldn't be registered.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app1}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Update app2 labels so it matches.
+	app2.SetStaticLabels(map[string]string{"group": "a", types.OriginLabel: types.OriginDynamic})
+	err = s.authServer.AuthServer.UpdateApp(ctx, app2)
+	require.NoError(t, err)
+
+	// Both should be registered now.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app1, app2}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Update app2 URI so it gets re-registered.
+	app2.SetURI("localhost:2345")
+	err = s.authServer.AuthServer.UpdateApp(ctx, app2)
+	require.NoError(t, err)
+
+	// app2 should get updated.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app1, app2}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Update app1 labels so it doesn't match.
+	app1.SetStaticLabels(map[string]string{"group": "c", types.OriginLabel: types.OriginDynamic})
+	err = s.authServer.AuthServer.UpdateApp(ctx, app1)
+	require.NoError(t, err)
+
+	// Only app0 and app2 should remain registered.
+	select {
+	case a := <-reconcileCh:
+		sort.Sort(a)
+		require.Empty(t, cmp.Diff(types.Apps{app0, app2}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+
+	// Remove app2.
+	err = s.authServer.AuthServer.DeleteApp(ctx, app2.GetName())
+	require.NoError(t, err)
+
+	// Only static app should remain.
+	select {
+	case a := <-reconcileCh:
+		require.Empty(t, cmp.Diff(types.Apps{app0}, a,
+			cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+		))
+	case <-time.After(time.Second):
+		t.Fatal("Didn't receive reconcile event after 1s.")
+	}
+}
+
+func makeStaticApp(name string, labels map[string]string) (*types.AppV3, error) {
+	return makeApp(name, labels, map[string]string{
+		types.OriginLabel: types.OriginConfigFile,
+	})
+}
+
+func makeDynamicApp(name string, labels map[string]string) (*types.AppV3, error) {
+	return makeApp(name, labels, map[string]string{
+		types.OriginLabel: types.OriginDynamic,
+	})
+}
+
+func makeApp(name string, labels map[string]string, additionalLabels map[string]string) (*types.AppV3, error) {
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	for k, v := range additionalLabels {
+		labels[k] = v
+	}
+	return types.NewAppV3(types.Metadata{
+		Name:   name,
+		Labels: labels,
+	}, types.AppSpecV3{
+		URI: "localhost",
+	})
+}

lib/srv/db/server.go

@@ -166,6 +166,8 @@ type Server struct {
 	mu sync.RWMutex
 	// log is used for logging.
 	log *logrus.Entry
+	// reconciler is used to reconcile proxied databases with database resources.
+	reconciler *services.Reconciler
 }
 
 // New returns a new database server.
@@ -190,6 +192,12 @@ func New(ctx context.Context, config Config) (*Server, error) {
 		},
 	}
 
+	// Reconciler will be reconciling database resources with proxied databases.
+	server.reconciler, err = server.getReconciler()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	// Update TLS config to require client certificate.
 	server.cfg.TLSConfig.ClientAuth = tls.RequireAndVerifyClientCert
 	server.cfg.TLSConfig.GetConfigForClient = getConfigForClient(
@@ -298,8 +306,8 @@ func (s *Server) registerDatabase(ctx context.Context, database types.Database)
 	return nil
 }
 
-// reRegisterDatabase updates database that is already registered.
-func (s *Server) reRegisterDatabase(ctx context.Context, database types.Database) error {
+// updateDatabase updates database that is already registered.
+func (s *Server) updateDatabase(ctx context.Context, database types.Database) error {
 	// Stop heartbeat and dynamic labels before starting new ones.
 	if err := s.stopDatabase(ctx, database.GetName()); err != nil {
 		return trace.Wrap(err)
@@ -442,7 +450,7 @@ func (s *Server) Start(ctx context.Context) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
-	if err := s.reconcileResources(ctx, databases); err != nil {
+	if err := s.reconciler.Reconcile(ctx, types.Databases(databases).AsResources()); err != nil {
 		return trace.Wrap(err)
 	}
 

lib/srv/db/watcher.go

@@ -29,16 +29,15 @@ import (
 // startWatcher starts watching changes to database resources and registers /
 // unregisters the proxied databases accordingly.
 func (s *Server) startWatcher(ctx context.Context) (*services.DatabaseWatcher, error) {
-	log := s.log.WithField("selector", s.cfg.Selectors)
 	if len(s.cfg.Selectors) == 0 {
-		log.Debug("Not initializing database resource watcher.")
+		s.log.Debug("Not initializing database resource watcher.")
 		return nil, nil
 	}
-	log.Debug("Initializing database resource watcher.")
+	s.log.Debug("Initializing database resource watcher.")
 	watcher, err := services.NewDatabaseWatcher(ctx, services.DatabaseWatcherConfig{
 		ResourceWatcherConfig: services.ResourceWatcherConfig{
 			Component: teleport.ComponentDatabase,
-			Log:       log,
+			Log:       s.log,
 			Client:    s.cfg.AccessPoint,
 		},
 	})
@@ -50,13 +49,13 @@ func (s *Server) startWatcher(ctx context.Context) (*services.DatabaseWatcher, e
 		for {
 			select {
 			case databases := <-watcher.DatabasesC:
-				if err := s.reconcileResources(ctx, databases); err != nil {
-					log.WithError(err).Errorf("Failed to reconcile %v.", databases)
+				if err := s.reconciler.Reconcile(ctx, databases.AsResources()); err != nil {
+					s.log.WithError(err).Errorf("Failed to reconcile %v.", databases)
 				} else if s.cfg.OnReconcile != nil {
 					s.cfg.OnReconcile(s.getDatabases())
 				}
 			case <-ctx.Done():
-				log.Debug("Database resource watcher done.")
+				s.log.Debug("Database resource watcher done.")
 				return
 			}
 		}
@@ -64,57 +63,37 @@ func (s *Server) startWatcher(ctx context.Context) (*services.DatabaseWatcher, e
 	return watcher, nil
 }
 
-// reconcileResources reconciles the database resources this server is currently
-// proxying with the provided up-to-date list of cluster databases.
-func (s *Server) reconcileResources(ctx context.Context, newResources types.Databases) error {
-	s.log.Debugf("Reconciling with %v database resources.", len(newResources))
-	var errs []error
-	// First remove databases that aren't present in the resource list anymore.
-	for _, current := range s.getDatabases() {
-		// Skip databases from static configuration. For backwards compatibility
-		// also consider empty "origin" value.
-		if current.Origin() == types.OriginConfigFile || current.Origin() == "" {
-			continue
-		}
-		if new := newResources.Find(current.GetName()); new == nil {
-			s.log.Infof("%v removed, unregistering.", current)
-			if err := s.unregisterDatabase(ctx, current.GetName()); err != nil {
-				errs = append(errs, trace.Wrap(err, "failed to unregister %v", current))
-			}
-		}
+func (s *Server) getReconciler() (*services.Reconciler, error) {
+	return services.NewReconciler(services.ReconcilerConfig{
+		Selectors:    s.cfg.Selectors,
+		GetResources: s.getResources,
+		OnCreate:     s.onCreate,
+		OnUpdate:     s.onUpdate,
+		OnDelete:     s.onDelete,
+		Log:          s.log,
+	})
+}
+
+func (s *Server) getResources() (resources types.ResourcesWithLabels) {
+	return s.getDatabases().AsResources()
+}
+
+func (s *Server) onCreate(ctx context.Context, resource types.ResourceWithLabels) error {
+	database, ok := resource.(types.Database)
+	if !ok {
+		return trace.BadParameter("expected types.Database, got %T", resource)
 	}
-	// Then add new databases if there are any or refresh those that were updated.
-	for _, new := range newResources {
-		if db := s.getDatabases().Find(new.GetName()); db == nil {
-			// Database with this name isn't registered yet, see if it matches.
-			if services.MatchDatabase(s.cfg.Selectors, new) {
-				s.log.Infof("%v matches, registering.", new)
-				if err := s.registerDatabase(ctx, new); err != nil {
-					errs = append(errs, trace.Wrap(err, "failed to register %v", new))
-				}
-			} else {
-				s.log.Debugf("%v doesn't match, not registering.", new)
-			}
-		} else if db.Origin() == types.OriginConfigFile || db.Origin() == "" {
-			// Do not overwrite databases from static configuration.
-			s.log.Infof("%v is part of static configuration, not registering %v.", db, new)
-			continue
-		} else if new.GetResourceID() != db.GetResourceID() {
-			// If labels were updated, the database may no longer match.
-			if services.MatchDatabase(s.cfg.Selectors, new) {
-				s.log.Infof("%v updated, re-registering.", new)
-				if err := s.reRegisterDatabase(ctx, new); err != nil {
-					errs = append(errs, trace.Wrap(err, "failed to re-register %v", new))
-				}
-			} else {
-				s.log.Infof("%v updated and no longer matches, unregistering.", new)
-				if err := s.unregisterDatabase(ctx, new.GetName()); err != nil {
-					errs = append(errs, trace.Wrap(err, "failed to unregister %v", new))
-				}
-			}
-		} else {
-			s.log.Debugf("%v is already registered.", new)
-		}
+	return s.registerDatabase(ctx, database)
+}
+
+func (s *Server) onUpdate(ctx context.Context, resource types.ResourceWithLabels) error {
+	database, ok := resource.(types.Database)
+	if !ok {
+		return trace.BadParameter("expected types.Database, got %T", resource)
 	}
-	return trace.NewAggregate(errs...)
+	return s.updateDatabase(ctx, database)
+}
+
+func (s *Server) onDelete(ctx context.Context, resource types.ResourceWithLabels) error {
+	return s.unregisterDatabase(ctx, resource.GetName())
 }