Activate direct reconciliation (only) with a feature flag

This enables us to merge the code without activating it yet, and test
it against the old code.

Current syntax is:
`export KCC_USE_DIRECT_RECONCILERS=APIKeysKey,<kind2>`

However I think we should consider this syntax provisional and subject
to change.

pkg/controller/registration/registration_controller.go

@@ -34,17 +34,19 @@ import (
 	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/crd/crdgeneration"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/dcl/conversion"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/k8s"
+	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/kccfeatureflags"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/servicemapping/servicemappingloader"
 
 	"github.com/GoogleCloudPlatform/declarative-resource-client-library/dcl"
 	tfschema "github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	apiextensions "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	crcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-	klog "sigs.k8s.io/controller-runtime/pkg/log"
+	crlog "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -54,7 +56,7 @@ const controllerName = "registration-controller"
 const serviceAccountKeyAPIGroup = "iam.cnrm.cloud.google.com"
 const serviceAccountKeyKind = "IAMServiceAccountKey"
 
-var logger = klog.Log.WithName(controllerName)
+var logger = crlog.Log.WithName(controllerName)
 
 // Add creates a new registration Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
@@ -170,8 +172,22 @@ func registerDefaultController(r *ReconcileRegistration, config *controller.Conf
 	if _, ok := k8s.IgnoredKindList[crd.Spec.Names.Kind]; ok {
 		return nil, nil
 	}
-	// Depending on which resource it is, we need to register a different controller.
+
 	var schemaUpdater k8s.SchemaReferenceUpdater
+
+	if kccfeatureflags.UseDirectReconciler(gvk.GroupKind()) {
+		switch gvk.GroupKind() {
+		case schema.GroupKind{Group: "apikeys.cnrm.cloud.google.com", Kind: "APIKeysKey"}:
+			if err := apikeys.AddKeyReconciler(r.mgr, config); err != nil {
+				return nil, err
+			}
+			return schemaUpdater, nil
+		default:
+			klog.Warningf("requested direct reconciler for %v, but it is not supported", gvk.GroupKind())
+		}
+	}
+
+	// Depending on which resource it is, we need to register a different controller.
 	switch gvk.Kind {
 	case "IAMPolicy":
 		if err := policy.Add(r.mgr, r.provider, r.smLoader, r.dclConverter, r.dclConfig, r.defaulters); err != nil {
@@ -189,10 +205,7 @@ func registerDefaultController(r *ReconcileRegistration, config *controller.Conf
 		if err := auditconfig.Add(r.mgr, r.provider, r.smLoader, r.dclConverter, r.dclConfig, r.defaulters); err != nil {
 			return nil, err
 		}
-	case "APIKeysKey":
-		if err := apikeys.AddKeyReconciler(r.mgr, config); err != nil {
-			return nil, err
-		}
+
 	default:
 		// register controllers for dcl-based CRDs
 		if val, ok := crd.Labels[k8s.DCL2CRDLabel]; ok && val == "true" {

pkg/kccfeatureflags/flags.go

@@ -0,0 +1,38 @@
+// Copyright 2024 Google LLC
+//
+// 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 kccfeatureflags
+
+import (
+	"os"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// UseDirectReconciler is true if we should use the direct reconciler to actuate the specified resource.
+func UseDirectReconciler(gk schema.GroupKind) bool {
+	directReconcilers := os.Getenv("KCC_USE_DIRECT_RECONCILERS")
+	if directReconcilers == "" {
+		return false
+	}
+
+	for _, directReconciler := range strings.Split(directReconcilers, ",") {
+		if directReconciler == gk.Kind {
+			return true
+		}
+	}
+
+	return false
+}