Merge pull request #683 from skriss/fail-fast-if-missing-crds

exit server if not all Ark CRDs exist at startup

pkg/apis/ark/v1/register.go

@@ -41,27 +41,41 @@ func Resource(resource string) schema.GroupResource {
 	return SchemeGroupVersion.WithResource(resource).GroupResource()
 }
 
+type typeInfo struct {
+	PluralName   string
+	ItemType     runtime.Object
+	ItemListType runtime.Object
+}
+
+func newTypeInfo(pluralName string, itemType, itemListType runtime.Object) typeInfo {
+	return typeInfo{
+		PluralName:   pluralName,
+		ItemType:     itemType,
+		ItemListType: itemListType,
+	}
+}
+
+// CustomResources returns a map of all custom resources within the Ark
+// API group, keyed on Kind.
+func CustomResources() map[string]typeInfo {
+	return map[string]typeInfo{
+		"Backup":              newTypeInfo("backups", &Backup{}, &BackupList{}),
+		"Restore":             newTypeInfo("restores", &Restore{}, &RestoreList{}),
+		"Schedule":            newTypeInfo("schedules", &Schedule{}, &ScheduleList{}),
+		"Config":              newTypeInfo("configs", &Config{}, &ConfigList{}),
+		"DownloadRequest":     newTypeInfo("downloadrequests", &DownloadRequest{}, &DownloadRequestList{}),
+		"DeleteBackupRequest": newTypeInfo("deletebackuprequests", &DeleteBackupRequest{}, &DeleteBackupRequestList{}),
+		"PodVolumeBackup":     newTypeInfo("podvolumebackups", &PodVolumeBackup{}, &PodVolumeBackupList{}),
+		"PodVolumeRestore":    newTypeInfo("podvolumerestores", &PodVolumeRestore{}, &PodVolumeRestoreList{}),
+		"ResticRepository":    newTypeInfo("resticrepositories", &ResticRepository{}, &ResticRepositoryList{}),
+	}
+}
+
 func addKnownTypes(scheme *runtime.Scheme) error {
-	scheme.AddKnownTypes(SchemeGroupVersion,
-		&Backup{},
-		&BackupList{},
-		&Schedule{},
-		&ScheduleList{},
-		&Restore{},
-		&RestoreList{},
-		&Config{},
-		&ConfigList{},
-		&DownloadRequest{},
-		&DownloadRequestList{},
-		&DeleteBackupRequest{},
-		&DeleteBackupRequestList{},
-		&PodVolumeBackup{},
-		&PodVolumeBackupList{},
-		&PodVolumeRestore{},
-		&PodVolumeRestoreList{},
-		&ResticRepository{},
-		&ResticRepositoryList{},
-	)
+	for _, typeInfo := range CustomResources() {
+		scheme.AddKnownTypes(SchemeGroupVersion, typeInfo.ItemType, typeInfo.ItemListType)
+	}
+
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil
 }

pkg/cmd/server/server.go

@@ -38,6 +38,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
@@ -150,6 +152,7 @@ type server struct {
 	backupService         cloudprovider.BackupService
 	snapshotService       cloudprovider.SnapshotService
 	discoveryClient       discovery.DiscoveryInterface
+	discoveryHelper       arkdiscovery.Helper
 	dynamicClient         dynamic.Interface
 	sharedInformerFactory informers.SharedInformerFactory
 	ctx                   context.Context
@@ -218,6 +221,15 @@ func (s *server) run() error {
 		return err
 	}
 
+	if err := s.initDiscoveryHelper(); err != nil {
+		return err
+	}
+
+	// check to ensure all Ark CRDs exist
+	if err := s.arkResourcesExist(); err != nil {
+		return err
+	}
+
 	originalConfig, err := s.loadConfig()
 	if err != nil {
 		return err
@@ -264,6 +276,68 @@ func (s *server) namespaceExists(namespace string) error {
 	return nil
 }
 
+// initDiscoveryHelper instantiates the server's discovery helper and spawns a
+// goroutine to call Refresh() every 5 minutes.
+func (s *server) initDiscoveryHelper() error {
+	discoveryHelper, err := arkdiscovery.NewHelper(s.discoveryClient, s.logger)
+	if err != nil {
+		return err
+	}
+	s.discoveryHelper = discoveryHelper
+
+	go wait.Until(
+		func() {
+			if err := discoveryHelper.Refresh(); err != nil {
+				s.logger.WithError(err).Error("Error refreshing discovery")
+			}
+		},
+		5*time.Minute,
+		s.ctx.Done(),
+	)
+
+	return nil
+}
+
+// arkResourcesExist checks for the existence of each Ark CRD via discovery
+// and returns an error if any of them don't exist.
+func (s *server) arkResourcesExist() error {
+	s.logger.Info("Checking existence of Ark custom resource definitions")
+
+	var arkGroupVersion *metav1.APIResourceList
+	for _, gv := range s.discoveryHelper.Resources() {
+		if gv.GroupVersion == api.SchemeGroupVersion.String() {
+			arkGroupVersion = gv
+			break
+		}
+	}
+
+	if arkGroupVersion == nil {
+		return errors.Errorf("Ark API group %s not found", api.SchemeGroupVersion)
+	}
+
+	foundResources := sets.NewString()
+	for _, resource := range arkGroupVersion.APIResources {
+		foundResources.Insert(resource.Kind)
+	}
+
+	var errs []error
+	for kind := range api.CustomResources() {
+		if foundResources.Has(kind) {
+			s.logger.WithField("kind", kind).Debug("Found custom resource")
+			continue
+		}
+
+		errs = append(errs, errors.Errorf("custom resource %s not found in Ark API group %s", kind, api.SchemeGroupVersion))
+	}
+
+	if len(errs) > 0 {
+		return kubeerrs.NewAggregate(errs)
+	}
+
+	s.logger.Info("All Ark custom resource definitions exist")
+	return nil
+}
+
 func (s *server) loadConfig() (*api.Config, error) {
 	s.logger.Info("Retrieving Ark configuration")
 	var (
@@ -542,27 +616,13 @@ func (s *server) runControllers(config *api.Config) error {
 		wg.Done()
 	}()
 
-	discoveryHelper, err := arkdiscovery.NewHelper(s.discoveryClient, s.logger)
-	if err != nil {
-		return err
-	}
-	go wait.Until(
-		func() {
-			if err := discoveryHelper.Refresh(); err != nil {
-				s.logger.WithError(err).Error("Error refreshing discovery")
-			}
-		},
-		5*time.Minute,
-		ctx.Done(),
-	)
-
 	if config.RestoreOnlyMode {
 		s.logger.Info("Restore only mode - not starting the backup, schedule, delete-backup, or GC controllers")
 	} else {
 		backupTracker := controller.NewBackupTracker()
 
 		backupper, err := backup.NewKubernetesBackupper(
-			discoveryHelper,
+			s.discoveryHelper,
 			client.NewDynamicFactory(s.dynamicClient),
 			podexec.NewPodCommandExecutor(s.kubeClientConfig, s.kubeClient.CoreV1().RESTClient()),
 			s.snapshotService,
@@ -638,7 +698,7 @@ func (s *server) runControllers(config *api.Config) error {
 	}
 
 	restorer, err := restore.NewKubernetesRestorer(
-		discoveryHelper,
+		s.discoveryHelper,
 		client.NewDynamicFactory(s.dynamicClient),
 		s.backupService,
 		s.snapshotService,

pkg/cmd/server/server_test.go

@@ -22,6 +22,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/heptio/ark/pkg/apis/ark/v1"
 	arktest "github.com/heptio/ark/pkg/util/test"
 )
@@ -51,3 +53,47 @@ func TestApplyConfigDefaults(t *testing.T) {
 	assert.Equal(t, 3*time.Minute, c.ScheduleSyncPeriod.Duration)
 	assert.Equal(t, []string{"a", "b"}, c.ResourcePriorities)
 }
+
+func TestArkResourcesExist(t *testing.T) {
+	var (
+		fakeDiscoveryHelper = &arktest.FakeDiscoveryHelper{}
+		server              = &server{
+			logger:          arktest.NewLogger(),
+			discoveryHelper: fakeDiscoveryHelper,
+		}
+	)
+
+	// Ark API group doesn't exist in discovery: should error
+	fakeDiscoveryHelper.ResourceList = []*metav1.APIResourceList{
+		{
+			GroupVersion: "foo/v1",
+			APIResources: []metav1.APIResource{
+				{
+					Name: "Backups",
+					Kind: "Backup",
+				},
+			},
+		},
+	}
+	assert.Error(t, server.arkResourcesExist())
+
+	// Ark API group doesn't contain any custom resources: should error
+	arkAPIResourceList := &metav1.APIResourceList{
+		GroupVersion: v1.SchemeGroupVersion.String(),
+	}
+
+	fakeDiscoveryHelper.ResourceList = append(fakeDiscoveryHelper.ResourceList, arkAPIResourceList)
+	assert.Error(t, server.arkResourcesExist())
+
+	// Ark API group contains all custom resources: should not error
+	for kind := range v1.CustomResources() {
+		arkAPIResourceList.APIResources = append(arkAPIResourceList.APIResources, metav1.APIResource{
+			Kind: kind,
+		})
+	}
+	assert.NoError(t, server.arkResourcesExist())
+
+	// Ark API group contains some but not all custom resources: should error
+	arkAPIResourceList.APIResources = arkAPIResourceList.APIResources[:3]
+	assert.Error(t, server.arkResourcesExist())
+}

pkg/install/crd.go

@@ -27,17 +27,13 @@ import (
 
 // CRDs returns a list of the CRD types for all of the required Ark CRDs
 func CRDs() []*apiextv1beta1.CustomResourceDefinition {
-	return []*apiextv1beta1.CustomResourceDefinition{
-		crd("Backup", "backups"),
-		crd("Schedule", "schedules"),
-		crd("Restore", "restores"),
-		crd("Config", "configs"),
-		crd("DownloadRequest", "downloadrequests"),
-		crd("DeleteBackupRequest", "deletebackuprequests"),
-		crd("PodVolumeBackup", "podvolumebackups"),
-		crd("PodVolumeRestore", "podvolumerestores"),
-		crd("ResticRepository", "resticrepositories"),
+	var crds []*apiextv1beta1.CustomResourceDefinition
+
+	for kind, typeInfo := range arkv1.CustomResources() {
+		crds = append(crds, crd(kind, typeInfo.PluralName))
 	}
+
+	return crds
 }
 
 func crd(kind, plural string) *apiextv1beta1.CustomResourceDefinition {