Enrollment controller updates to use instance Properties as key (#701)

Signed-off-by: Steven Kaufer <[email protected]>

Author: kaufers

pkg/controller/enrollment/enroller.go

@@ -46,6 +46,8 @@ type enroller struct {
 
 	// template that we use to render with a source instance.Description to get the link Key
 	sourceKeySelectorTemplate *template.Template
+	// template that we use to render with an enrollment instance.Description to get the link Key
+	enrollmentKeySelectorTemplate *template.Template
 	// template used to render the enrollment's Provision propertiesx
 	enrollmentPropertiesTemplate *template.Template
 }

pkg/controller/enrollment/enroller_test.go

@@ -139,3 +139,224 @@ options:
 		"Destroy",
 	}, <-seen)
 }
+
+func TestEnrollerNoTags(t *testing.T) {
+
+	// Group members: 1, 2, 3
+	source := []instance.Description{
+		{ID: instance.ID("instance-1"), Properties: types.AnyString(`{"backend_id":"1"}`)},
+		{ID: instance.ID("instance-2"), Properties: types.AnyString(`{"backend_id":"2"}`)},
+		{ID: instance.ID("instance-3"), Properties: types.AnyString(`{"backend_id":"3"}`)},
+	}
+
+	// Currently enrolled: 1, 2, 4
+	enrolled := []instance.Description{
+		{ID: instance.ID("1")},
+		{ID: instance.ID("2")},
+		{ID: instance.ID("4")},
+	}
+
+	seen := make(chan []interface{}, 10)
+
+	enroller := newEnroller(
+		func() discovery.Plugins {
+			return fakePlugins{
+				"test": &plugin.Endpoint{},
+			}
+		},
+		fakeLeader(func() (bool, error) { return false, nil }),
+		enrollment.Options{})
+	enroller.groupPlugin = &group_test.Plugin{
+		DoDescribeGroup: func(gid group.ID) (group.Description, error) {
+			result := group.Description{Instances: source}
+			return result, nil
+		},
+	}
+	enroller.instancePlugin = &instance_test.Plugin{
+		DoDescribeInstances: func(t map[string]string, p bool) ([]instance.Description, error) {
+			return enrolled, nil
+		},
+		DoProvision: func(spec instance.Spec) (*instance.ID, error) {
+
+			seen <- []interface{}{spec, "Provision"}
+			return nil, nil
+		},
+		DoDestroy: func(id instance.ID, ctx instance.Context) error {
+
+			seen <- []interface{}{id, ctx, "Destroy"}
+			return nil
+		},
+	}
+
+	require.False(t, enroller.Running())
+
+	// Build a spec that uses the "backend_id" as the key for the source and just
+	// the "ID" for the enrolled
+	spec := types.Spec{}
+	require.NoError(t, types.AnyYAMLMust([]byte(`
+kind: enrollment
+metadata:
+  name: nfs
+properties:
+  List: group/workers
+  Instance:
+    Plugin: nfs/authorization
+    Properties:
+       backend_id: \{\{ $x := .Properties | jsonDecode \}\}\{\{ int $x.backend_id \}\}
+options:
+  SourceKeySelector: \{\{ $x := .Properties | jsonDecode \}\}\{\{ int $x.backend_id \}\}
+  EnrollmentKeySelector: \{\{.ID\}\}
+`)).Decode(&spec))
+
+	require.NoError(t, enroller.updateSpec(spec))
+
+	st, err := enroller.getSourceKeySelectorTemplate()
+	require.NoError(t, err)
+	require.NotNil(t, st)
+
+	et, err := enroller.getEnrollmentPropertiesTemplate()
+	require.NoError(t, err)
+	require.NotNil(t, et)
+
+	require.NoError(t, err)
+
+	s, err := enroller.getSourceInstances()
+	require.NoError(t, err)
+	require.Equal(t, source, s)
+
+	found, err := enroller.getEnrolledInstances()
+	require.NoError(t, err)
+	require.Equal(t, enrolled, found)
+
+	require.NoError(t, enroller.sync())
+
+	// check the provision and destroy calls, instance 3 should be added and 4
+	// should be removed
+	require.Equal(t, []interface{}{
+		instance.Spec{
+			Properties: types.AnyString(`{"backend_id":"3"}`),
+			Tags: map[string]string{
+				"infrakit.enrollment.sourceID": "instance-3",
+				"infrakit.enrollment.name":     "nfs",
+			},
+		},
+		"Provision",
+	}, <-seen)
+	require.Equal(t, []interface{}{
+		instance.ID("4"),
+		instance.Termination,
+		"Destroy",
+	}, <-seen)
+	require.Len(t, seen, 0)
+}
+
+func TestEnrollerMissingProps(t *testing.T) {
+
+	// Group members: 1, 2, 3 (no props), 4
+	source := []instance.Description{
+		{ID: instance.ID("instance-1"), Properties: types.AnyString(`{"backend_id":"1"}`)},
+		{ID: instance.ID("instance-2"), Properties: types.AnyString(`{"backend_id":"2"}`)},
+		{ID: instance.ID("instance-3")},
+		{ID: instance.ID("instance-4"), Properties: types.AnyString(`{"backend_id":"4"}`)},
+	}
+
+	// Currently enrolled: 1, 2, 5
+	enrolled := []instance.Description{
+		{ID: instance.ID("1")},
+		{ID: instance.ID("2")},
+		{ID: instance.ID("5")},
+	}
+
+	seen := make(chan []interface{}, 10)
+
+	enroller := newEnroller(
+		func() discovery.Plugins {
+			return fakePlugins{
+				"test": &plugin.Endpoint{},
+			}
+		},
+		fakeLeader(func() (bool, error) { return false, nil }),
+		enrollment.Options{})
+	enroller.groupPlugin = &group_test.Plugin{
+		DoDescribeGroup: func(gid group.ID) (group.Description, error) {
+			result := group.Description{Instances: source}
+			return result, nil
+		},
+	}
+	enroller.instancePlugin = &instance_test.Plugin{
+		DoDescribeInstances: func(t map[string]string, p bool) ([]instance.Description, error) {
+			return enrolled, nil
+		},
+		DoProvision: func(spec instance.Spec) (*instance.ID, error) {
+
+			seen <- []interface{}{spec, "Provision"}
+			return nil, nil
+		},
+		DoDestroy: func(id instance.ID, ctx instance.Context) error {
+
+			seen <- []interface{}{id, ctx, "Destroy"}
+			return nil
+		},
+	}
+
+	require.False(t, enroller.Running())
+
+	// Build a spec that uses the "backend_id" as the key for the source and just
+	// the "ID" for the enrolled
+	spec := types.Spec{}
+	require.NoError(t, types.AnyYAMLMust([]byte(`
+kind: enrollment
+metadata:
+  name: nfs
+properties:
+  List: group/workers
+  Instance:
+    Plugin: nfs/authorization
+    Properties:
+       backend_id: \{\{ $x := .Properties | jsonDecode \}\}\{\{ int $x.backend_id \}\}
+options:
+  SourceKeySelector: \{\{ $x := .Properties | jsonDecode \}\}\{\{ int $x.backend_id \}\}
+  EnrollmentKeySelector: \{\{.ID\}\}
+`)).Decode(&spec))
+
+	require.NoError(t, enroller.updateSpec(spec))
+
+	st, err := enroller.getSourceKeySelectorTemplate()
+	require.NoError(t, err)
+	require.NotNil(t, st)
+
+	et, err := enroller.getEnrollmentPropertiesTemplate()
+	require.NoError(t, err)
+	require.NotNil(t, et)
+
+	require.NoError(t, err)
+
+	s, err := enroller.getSourceInstances()
+	require.NoError(t, err)
+	require.Equal(t, source, s)
+
+	found, err := enroller.getEnrolledInstances()
+	require.NoError(t, err)
+	require.Equal(t, enrolled, found)
+
+	require.NoError(t, enroller.sync())
+
+	// check the provision and destroy calls, instance 3 should be added, instance
+	// 4 should be ignored (cannot be indexed), and 5 should be removed
+	require.Equal(t, []interface{}{
+		instance.Spec{
+			Properties: types.AnyString(`{"backend_id":"4"}`),
+			Tags: map[string]string{
+				"infrakit.enrollment.sourceID": "instance-4",
+				"infrakit.enrollment.name":     "nfs",
+			},
+		},
+		"Provision",
+	}, <-seen)
+	require.Equal(t, []interface{}{
+		instance.ID("5"),
+		instance.Termination,
+		"Destroy",
+	}, <-seen)
+	require.Len(t, seen, 0)
+}

pkg/controller/enrollment/set.go

@@ -16,7 +16,7 @@ func index(list instance.Descriptions, getKey keyFunc) (map[string]instance.Desc
 	for _, n := range list {
 		key, err := getKey(n)
 		if err != nil {
-			log.Error("cannot index entry", "instane.Description", n)
+			log.Error("cannot index entry", "instance.Description", n, "err", err)
 			continue
 		}
 		this.Add(key)

pkg/controller/enrollment/sync.go

@@ -66,6 +66,23 @@ func (l *enroller) getSourceKeySelectorTemplate() (*template.Template, error) {
 	return l.sourceKeySelectorTemplate, nil
 }
 
+func (l *enroller) getEnrollmentKeySelectorTemplate() (*template.Template, error) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if l.options.EnrollmentKeySelector != "" {
+		if l.enrollmentKeySelectorTemplate == nil {
+			t, err := enrollment.TemplateFrom([]byte(l.options.EnrollmentKeySelector))
+			if err != nil {
+				return nil, err
+			}
+			l.enrollmentKeySelectorTemplate = t
+		}
+	}
+
+	return l.enrollmentKeySelectorTemplate, nil
+}
+
 func (l *enroller) getEnrollmentPropertiesTemplate() (*template.Template, error) {
 	l.lock.Lock()
 	defer l.lock.Unlock()
@@ -119,13 +136,27 @@ func (l *enroller) sync() error {
 		return string(d.ID), nil
 	}
 
-	// As long as the downstream enrollment records are labeled correctly we
-	// can even support 'importing' out-of-band created enrollment records
+	// If specified, use the given enrollment selectior to get the index key;
+	// else check for the labels so that we can even support 'importing'
+	// out-of-band created enrollment records
 	enrolledKeyFunc := func(d instance.Description) (string, error) {
-		if v, has := d.Tags["infrakit.enrollment.sourceID"]; has {
-			return v, nil
+
+		t, err := l.getEnrollmentKeySelectorTemplate()
+		if err != nil {
+			return "", err
+		}
+		if t == nil {
+			if v, has := d.Tags["infrakit.enrollment.sourceID"]; has {
+				return v, nil
+			}
+			return "", fmt.Errorf("not-matched:%v", d.ID)
 		}
-		return "", fmt.Errorf("not-matched:%v", d.ID)
+		view, err := t.Render(d)
+		if err != nil {
+			return "", err
+		}
+		return view, nil
+
 	}
 
 	// compute the delta required to make enrolled look like source

pkg/controller/enrollment/types/types.go

@@ -101,6 +101,10 @@ type Options struct {
 	// SourceKeySelector: \{\{ .ID \}\}  # selects the ID field.
 	SourceKeySelector string
 
+	// SourceKeySelector is a string template for selecting the join key from
+	// a enrollment plugin's instance.Description.
+	EnrollmentKeySelector string
+
 	// SyncInterval is the time interval between reconciliation. Syntax
 	// is go's time.Duration string representation (e.g. 1m, 30s)
 	SyncInterval types.Duration

pkg/plugin/group/scaled.go

@@ -133,7 +133,7 @@ func (s *scaledGroup) Destroy(inst instance.Description, ctx instance.Context) e
 func (s *scaledGroup) List() ([]instance.Description, error) {
 	settings := s.latestSettings()
 
-	return settings.instancePlugin.DescribeInstances(s.memberTags, false)
+	return settings.instancePlugin.DescribeInstances(s.memberTags, true)
 }
 
 func (s *scaledGroup) Label() error {