add initial sidecar Bucket creation reconciliation

Add initial sidecar implementation for dynamic Bucket provisioning.
The implementation is consistent with v1alpha2 pre-approval designs.
Translating driver-returned gRPC protocol information into user-facing
env var format is partially implemented. This is a core part of what
COSI needs to get right internally. Scaffolding is set in place, and
implementation for S3 translation is implemented. Azure and GCS
translations are left as to-do items.

Signed-off-by: Blaine Gardner <[email protected]>

Author: BlaineEXE

internal/predicate/predicate.go

@@ -66,11 +66,11 @@ func AnyGeneric() predicate.Funcs {
 // where the generation changes. For most resources, a generation change means that the resource
 // `spec` has changed, ignoring metadata and status changes.
 //
-// The predicate does not enqueue requests for any Create/Update/Delete events.
+// The predicate does not enqueue requests for any Create/Delete/Generic events.
 // This ensures that other predicates can effectively filter out undesired non-Update events.
 //
 // This is a modified implementation of controller-runtime's GenerationChangedPredicate{} which
-// does enqueue requests for all Create/Update/Delete events -- behavior COSI does not always want.
+// does enqueue requests for all Create/Delete/Generic events -- behavior COSI does not always want.
 func GenerationChangedInUpdateOnly() predicate.Funcs {
 	funcs := allFalseFuncs()
 	funcs.UpdateFunc = func(e event.UpdateEvent) bool {
@@ -83,7 +83,7 @@ func GenerationChangedInUpdateOnly() predicate.Funcs {
 // where the protection finalizer has been removed. This helps ensure that COSI always has a chance
 // to re-apply the protection finalizer when it's needed.
 //
-// The predicate does not enqueue requests for any Create/Update/Delete events.
+// The predicate does not enqueue requests for any Create/Delete/Generic events.
 // This ensures that other predicates can effectively filter out undesired non-Update events.
 func ProtectionFinalizerRemoved(s *runtime.Scheme) predicate.Funcs {
 	funcs := allFalseFuncs()

internal/protocol/azure.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+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 protocol
+
+import (
+	cosiapi "sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha2"
+	cosiproto "sigs.k8s.io/container-object-storage-interface/proto"
+)
+
+// AzureBucketInfoTranslator implements RpcApiTranslator for S3 bucket info.
+type AzureBucketInfoTranslator struct{}
+
+// TODO: S3CredentialTranslator implements RpcApiTranslator for S3 credentials.
+
+var _ RpcApiTranslator[*cosiproto.AzureBucketInfo, cosiapi.BucketInfoVar] = AzureBucketInfoTranslator{}
+
+func (AzureBucketInfoTranslator) RpcToApi(b *cosiproto.AzureBucketInfo) map[cosiapi.BucketInfoVar]string {
+	if b == nil {
+		return nil
+	}
+
+	out := map[cosiapi.BucketInfoVar]string{
+		cosiapi.BucketInfoVar_Azure_StorageAccount: "",
+	}
+
+	// TODO: implement
+
+	return out
+}
+
+func (AzureBucketInfoTranslator) ApiToRpc(vars map[cosiapi.BucketInfoVar]string) *cosiproto.AzureBucketInfo {
+	if len(vars) == 0 {
+		return nil
+	}
+
+	// TODO: implement
+
+	return nil
+}
+
+func (AzureBucketInfoTranslator) Validate(
+	vars map[cosiapi.BucketInfoVar]string, _ cosiapi.BucketAccessAuthenticationType,
+) error {
+	// TODO: implement
+
+	return nil
+}

internal/protocol/gcs.go

@@ -0,0 +1,62 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+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 protocol
+
+import (
+	cosiapi "sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha2"
+	cosiproto "sigs.k8s.io/container-object-storage-interface/proto"
+)
+
+// GcsBucketInfoTranslator implements RpcApiTranslator for S3 bucket info.
+type GcsBucketInfoTranslator struct{}
+
+// TODO: S3CredentialTranslator implements RpcApiTranslator for S3 credentials.
+
+var _ RpcApiTranslator[*cosiproto.GcsBucketInfo, cosiapi.BucketInfoVar] = GcsBucketInfoTranslator{}
+
+func (GcsBucketInfoTranslator) RpcToApi(b *cosiproto.GcsBucketInfo) map[cosiapi.BucketInfoVar]string {
+	if b == nil {
+		return nil
+	}
+
+	out := map[cosiapi.BucketInfoVar]string{
+		cosiapi.BucketInfoVar_GCS_BucketName: "",
+		cosiapi.BucketInfoVar_GCS_ProjectId:  "",
+	}
+
+	// TODO: implement
+
+	return out
+}
+
+func (GcsBucketInfoTranslator) ApiToRpc(vars map[cosiapi.BucketInfoVar]string) *cosiproto.GcsBucketInfo {
+	if len(vars) == 0 {
+		return nil
+	}
+
+	// TODO: implement
+
+	return nil
+}
+
+func (GcsBucketInfoTranslator) Validate(
+	vars map[cosiapi.BucketInfoVar]string, _ cosiapi.BucketAccessAuthenticationType,
+) error {
+	// TODO: implement
+
+	return nil
+}

internal/protocol/protocol.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+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 protocol contains definitions and functions for transforming COSI gRPC spec definitions
+// into COSI Kubernetes definitions.
+package protocol
+
+import (
+	"fmt"
+
+	cosiapi "sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha2"
+	cosiproto "sigs.k8s.io/container-object-storage-interface/proto"
+)
+
+// ObjectProtocolTranslator translates object protocol types between the RPC driver-domain and
+// Kubernetes API user-domain.
+type ObjectProtocolTranslator struct{}
+
+var (
+	objectProtocolProtoApiMapping = map[cosiproto.ObjectProtocol_Type]cosiapi.ObjectProtocol{
+		cosiproto.ObjectProtocol_S3:    cosiapi.ObjectProtocolS3,
+		cosiproto.ObjectProtocol_AZURE: cosiapi.ObjectProtocolAzure,
+		cosiproto.ObjectProtocol_GCS:   cosiapi.ObjectProtocolGcs,
+	}
+)
+
+// RpcToApi translates object protocols from RPC to API.
+func (ObjectProtocolTranslator) RpcToApi(in cosiproto.ObjectProtocol_Type) (cosiapi.ObjectProtocol, error) {
+	a, ok := objectProtocolProtoApiMapping[in]
+	if !ok {
+		return cosiapi.ObjectProtocol(""), fmt.Errorf("unknown driver protocol %q", string(in))
+	}
+	return a, nil
+}
+
+// ApiToRpc translates object protocols from API to RPC.
+func (ObjectProtocolTranslator) ApiToRpc(in cosiapi.ObjectProtocol) (cosiproto.ObjectProtocol_Type, error) {
+	for p, a := range objectProtocolProtoApiMapping {
+		if a == in {
+			return p, nil
+		}
+	}
+	return cosiproto.ObjectProtocol_UNKNOWN, fmt.Errorf("unknown api protocol %q", string(in))
+}
+
+// An RpcApiTranslator translates types between the RPC driver-domain and Kubernetes API user-domain
+// for a particular protocol.
+type RpcApiTranslator[RpcType any, ApiType comparable] interface {
+	// RpcToApi translates bucket info from RPC to API with no validation.
+	// If the input is nil, the result map MUST be nil.
+	// All possible API info fields SHOULD be present in the result, even if the corresponding RPC
+	// info is omitted; an empty string value should be used for keys with no explicit config.
+	RpcToApi(RpcType) map[ApiType]string
+
+	// ApiToRpc translates bucket info from API to RPC with no validation.
+	// If the input map is nil or empty, this is assumed to mean the protocol is not supported, and
+	// the result MUST be nil.
+	ApiToRpc(map[ApiType]string) RpcType
+
+	// Validate checks that user-domain API fields meet requirements and expectations.
+	Validate(map[ApiType]string, cosiapi.BucketAccessAuthenticationType) error
+}
+
+// contains is a helper that returns true if the given `list` contains the item `key`.
+// Useful for a variety of Validate() implementations.
+func contains[T comparable](list []T, key T) bool {
+	for _, i := range list {
+		if i == key {
+			return true
+		}
+	}
+	return false
+}

internal/protocol/protocol_test.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+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 protocol
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	cosiapi "sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha2"
+	cosiproto "sigs.k8s.io/container-object-storage-interface/proto"
+)
+
+func TestObjectProtocolTranslator(t *testing.T) {
+	tests := []struct {
+		name    string                 // description of this test case
+		api     cosiapi.ObjectProtocol // input
+		wantRpc cosiproto.ObjectProtocol_Type
+		wantErr bool
+	}{
+		{"empty string", cosiapi.ObjectProtocol(""), cosiproto.ObjectProtocol_UNKNOWN, true},
+		{"unknown proto", cosiapi.ObjectProtocol("evilcorp-proto"), cosiproto.ObjectProtocol_UNKNOWN, true},
+		{"S3", cosiapi.ObjectProtocolS3, cosiproto.ObjectProtocol_S3, false},
+		{"Azure", cosiapi.ObjectProtocolAzure, cosiproto.ObjectProtocol_AZURE, false},
+		{"GCS", cosiapi.ObjectProtocolGcs, cosiproto.ObjectProtocol_GCS, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			o := ObjectProtocolTranslator{}
+
+			gotRpc, gotErr := o.ApiToRpc(tt.api)
+			if tt.wantErr {
+				assert.Error(t, gotErr)
+			} else {
+				assert.NoError(t, gotErr)
+			}
+			assert.Equal(t, tt.wantRpc, gotRpc)
+
+			// Test the round trip. Tests UNKNOWN proto as input for tests where err is expected.
+			rtGot, rtErr := o.RpcToApi(tt.wantRpc)
+			if tt.wantErr {
+				assert.Error(t, rtErr)
+				assert.Equal(t, cosiapi.ObjectProtocol(""), rtGot)
+			} else {
+				assert.NoError(t, rtErr)
+				assert.Equal(t, tt.api, rtGot)
+			}
+		})
+	}
+}