WIP: Implement direct controller for Spanner Database

config/crds/resources/apiextensions.k8s.io_v1_customresourcedefinition_spannerdatabases.spanner.cnrm.cloud.google.com.yaml

@@ -8,7 +8,6 @@ metadata:
     cnrm.cloud.google.com/managed-by-kcc: "true"
     cnrm.cloud.google.com/stability-level: stable
     cnrm.cloud.google.com/system: "true"
-    cnrm.cloud.google.com/tf2crd: "true"
   name: spannerdatabases.spanner.cnrm.cloud.google.com
 spec:
   group: spanner.cnrm.cloud.google.com

go.mod

@@ -11,6 +11,7 @@ require (
 	cloud.google.com/go/monitoring v1.19.0
 	cloud.google.com/go/profiler v0.1.0
 	cloud.google.com/go/resourcemanager v1.9.7
+	cloud.google.com/go/spanner v1.60.0
 	contrib.go.opencensus.io/exporter/prometheus v0.1.0
 	github.com/GoogleCloudPlatform/declarative-resource-client-library v1.62.0
 	github.com/GoogleCloudPlatform/k8s-config-connector/mockgcp v0.0.0-00010101000000-000000000000

hack/compare-mock

@@ -5,6 +5,8 @@ set -x
 
 export KUBEBUILDER_ASSETS=$(go run sigs.k8s.io/controller-runtime/tools/setup-envtest@latest use -p path)
 
+export KCC_USE_DIRECT_RECONCILERS=SpannerDatabase
+
 rm -rf $(pwd)/artifactz/mocks
 
 RUN_TESTS=TestAllInSeries/$1

pkg/controller/direct/directbase/utils.go

@@ -0,0 +1,51 @@
+// 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 directbase
+
+import (
+	"errors"
+
+	"github.com/googleapis/gax-go/v2/apierror"
+	"k8s.io/klog/v2"
+)
+
+func ValueOf[T any](p *T) T {
+	var v T
+	if p != nil {
+		v = *p
+	}
+	return v
+}
+
+// HasHTTPCode returns true if the given error is an HTTP response with the given code.
+func HasHTTPCode(err error, code int) bool {
+	if err == nil {
+		return false
+	}
+	apiError := &apierror.APIError{}
+	if errors.As(err, &apiError) {
+		if apiError.HTTPCode() == code {
+			return true
+		}
+	} else {
+		klog.Warningf("unexpected error type %T", err)
+	}
+	return false
+}
+
+// IsNotFound returns true if the given error is an HTTP 404.
+func IsNotFound(err error) bool {
+	return HasHTTPCode(err, 404)
+}

pkg/controller/direct/register/register.go

@@ -20,4 +20,5 @@ import (
 	_ "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller/direct/gkehub"
 	_ "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller/direct/logging"
 	_ "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller/direct/resourcemanager"
+	_ "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller/direct/spanner"
 )

pkg/controller/direct/spanner/database_controller.go

@@ -0,0 +1,235 @@
+// 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 spanner
+
+import (
+	"context"
+	"fmt"
+
+	databaseapi "cloud.google.com/go/spanner/admin/database/apiv1"
+	"cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
+	"google.golang.org/api/option"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	krm "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/spanner/v1beta1"
+	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller"
+	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/controller/direct/directbase"
+	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/k8s"
+)
+
+const ctlrName = "spannerdatabase-controller"
+
+func init() {
+	directbase.ControllerBuilder.RegisterModel(krm.SpannerDatabaseGVK, NewSpannerDatabaseModel)
+}
+
+type spannerDatabaseModel struct {
+	config *controller.Config
+}
+
+var _ directbase.Model = &spannerDatabaseModel{}
+
+func NewSpannerDatabaseModel(config *controller.Config) directbase.Model {
+	return &spannerDatabaseModel{config: config}
+}
+
+var _ directbase.Adapter = &spannerDatabaseAdapter{}
+
+type spannerDatabaseAdapter struct {
+	projectID  string
+	instanceID string
+	databaseID string
+
+	desired *krm.SpannerDatabase
+
+	dbClient *databaseapi.DatabaseAdminClient
+}
+
+func (m *spannerDatabaseModel) client(ctx context.Context) (*databaseapi.DatabaseAdminClient, error) {
+	var opts []option.ClientOption
+	if m.config.UserAgent != "" {
+		opts = append(opts, option.WithUserAgent(m.config.UserAgent))
+	}
+	if m.config.HTTPClient != nil {
+		opts = append(opts, option.WithHTTPClient(m.config.HTTPClient))
+	}
+	if m.config.UserProjectOverride && m.config.BillingProject != "" {
+		opts = append(opts, option.WithQuotaProject(m.config.BillingProject))
+	}
+
+	gcpClient, err := databaseapi.NewDatabaseAdminRESTClient(ctx, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("building SpannerDatabase client: %w", err)
+	}
+	return gcpClient, err
+}
+
+// AdapterForObject implements the Model interface.
+func (m *spannerDatabaseModel) AdapterForObject(ctx context.Context, reader client.Reader, u *unstructured.Unstructured) (directbase.Adapter, error) {
+	client, err := m.client(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	obj := &krm.SpannerDatabase{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, &obj); err != nil {
+		return nil, fmt.Errorf("error converting to %T: %w", obj, err)
+	}
+
+	// TODO: Resolve external
+	instanceID := obj.Spec.InstanceRef.Name
+
+	// TODO(yuwenma): following current behavior. But do we have better option?
+	databaseID := directbase.ValueOf(obj.Spec.ResourceID)
+	if databaseID == "" {
+		databaseID = obj.GetName()
+	}
+
+	// TODO(yuwenma): following current behavior. But do we have better option?
+	projectID, ok := u.GetAnnotations()[k8s.ProjectIDAnnotation]
+	if !ok {
+		projectID = u.GetNamespace()
+	}
+
+	return &spannerDatabaseAdapter{
+		projectID:  projectID,
+		instanceID: instanceID,
+		databaseID: databaseID,
+		desired:    obj,
+		dbClient:   client,
+	}, nil
+}
+
+// Find implements the Adapter interface.
+func (a *spannerDatabaseAdapter) Find(ctx context.Context) (bool, error) {
+	if a.databaseID == "" {
+		return false, nil
+	}
+
+	req := &databasepb.GetDatabaseRequest{
+		Name: a.fullyQualifiedName(),
+	}
+	_, err := a.dbClient.GetDatabase(ctx, req)
+	if err != nil {
+		if directbase.IsNotFound(err) {
+			klog.Warningf("SpannerDatabase was not found: %v", err)
+			return false, nil
+		}
+		return false, err
+	}
+
+	return true, nil
+}
+
+// Delete implements the Adapter interface.
+func (a *spannerDatabaseAdapter) Delete(ctx context.Context) (bool, error) {
+	// TODO: Delete via status selfLink
+	req := &databasepb.DropDatabaseRequest{
+		Database: a.fullyQualifiedName(),
+	}
+	if err := a.dbClient.DropDatabase(ctx, req); err != nil {
+		if directbase.IsNotFound(err) {
+			return false, nil
+		}
+		return false, fmt.Errorf("deleting key: %w", err)
+	}
+	return true, nil
+}
+
+// Create implements the Adapter interface.
+func (a *spannerDatabaseAdapter) Create(ctx context.Context, u *unstructured.Unstructured) error {
+	log := klog.FromContext(ctx)
+	log.V(2).Info("creating object", "u", u)
+
+	req, err := a.spannerDatabaseKRMToCreateDatabaseRequest(a.desired)
+	if err != nil {
+		return fmt.Errorf("convert SpannerDatabase KRM to CreateDatabaseRequest API: %w", err)
+	}
+
+	log.Info("creating spannerDatabase", "spannerDatabase", req)
+
+	op, err := a.dbClient.CreateDatabase(ctx, req)
+	if err != nil {
+		return fmt.Errorf("creating spannerDatabase: %w", err)
+	}
+	created, err := op.Wait(ctx)
+	if err != nil {
+		return fmt.Errorf("waiting for spannerDatabase creation: %w", err)
+	}
+	log.V(2).Info("created spannerDatabase", "spannerDatabase", created)
+
+	return nil
+}
+
+func (a *spannerDatabaseAdapter) Update(ctx context.Context, u *unstructured.Unstructured) error {
+	// TODO
+	return nil
+}
+
+func (a *spannerDatabaseAdapter) Export(ctx context.Context) (*unstructured.Unstructured, error) {
+	return nil, nil
+}
+
+func (a *spannerDatabaseAdapter) fullyQualifiedName() string {
+	return fmt.Sprintf("projects/%s/instances/%s/databases/%s", a.projectID, a.instanceID, a.databaseID)
+}
+
+func (a *spannerDatabaseAdapter) spannerDatabaseKRMToCreateDatabaseRequest(r *krm.SpannerDatabase) (*databasepb.CreateDatabaseRequest, error) {
+	// Default database dialect is GOOGLE_STANDARD_SQL
+	databaseDialect := databasepb.DatabaseDialect_DATABASE_DIALECT_UNSPECIFIED
+	if r.Spec.DatabaseDialect != nil {
+		if directbase.ValueOf(r.Spec.DatabaseDialect) == "GOOGLE_STANDARD_SQL" {
+			databaseDialect = databasepb.DatabaseDialect_GOOGLE_STANDARD_SQL
+		} else if directbase.ValueOf(r.Spec.DatabaseDialect) == "POSTGRESQL" {
+			databaseDialect = databasepb.DatabaseDialect_POSTGRESQL
+		} else {
+			return nil, fmt.Errorf("unsupported database dialect: %s", directbase.ValueOf(r.Spec.DatabaseDialect))
+		}
+	}
+
+	// For POSTGRESQL, database name must be enclosed in double quotes
+	createDelimiter := '`'
+	if databaseDialect == databasepb.DatabaseDialect_POSTGRESQL {
+		createDelimiter = '"'
+	}
+	createStatement := fmt.Sprintf(
+		"CREATE DATABASE %c%s%c",
+		createDelimiter,
+		a.databaseID,
+		createDelimiter,
+	)
+
+	// Add version retention period if specified
+	extraStatements := r.Spec.Ddl
+	if r.Spec.VersionRetentionPeriod != nil {
+		extraStatements = append(extraStatements, fmt.Sprintf(
+			"ALTER DATABASE %c%s%c SET OPTIONS (version_retention_period = '%s')",
+			createDelimiter,
+			a.databaseID,
+			createDelimiter,
+			directbase.ValueOf(r.Spec.VersionRetentionPeriod)),
+		)
+	}
+
+	return &databasepb.CreateDatabaseRequest{
+		Parent:          fmt.Sprintf("projects/%s/instances/%s", a.projectID, a.instanceID),
+		CreateStatement: createStatement,
+		ExtraStatements: extraStatements,
+		DatabaseDialect: databaseDialect,
+	}, nil
+}

pkg/test/resourcefixture/testdata/basic/spanner/v1beta1/spannerdatabase/_generated_object_spannerdatabase.golden.yaml

@@ -3,14 +3,11 @@ kind: SpannerDatabase
 metadata:
   annotations:
     cnrm.cloud.google.com/management-conflict-prevention-policy: none
-    cnrm.cloud.google.com/mutable-but-unreadable-fields: '{"spec":{"ddl":["CREATE
-      TABLE t1 (t1 INT64 NOT NULL,) PRIMARY KEY(t1)"]}}'
     cnrm.cloud.google.com/project-id: ${projectId}
-    cnrm.cloud.google.com/state-into-spec: merge
   finalizers:
   - cnrm.cloud.google.com/finalizer
   - cnrm.cloud.google.com/deletion-defender
-  generation: 2
+  generation: 1
   labels:
     cnrm-test: "true"
   name: spannerdatabase-test
@@ -20,12 +17,11 @@ spec:
   - CREATE TABLE t1 (t1 INT64 NOT NULL,) PRIMARY KEY(t1)
   instanceRef:
     name: spannerinstance-${uniqueId}
-  resourceID: spannerdatabase-test
 status:
   conditions:
   - lastTransitionTime: "1970-01-01T00:00:00Z"
     message: The resource is up to date
     reason: UpToDate
     status: "True"
     type: Ready
-  observedGeneration: 2
+  observedGeneration: 1