Add configuration for data residency (#1681)

* add configuration map for data residency

* Use array instead of csv string for allowedpersistenceregions configuration

* Add default to the data residency config

* Add TestNewDefaultsConfigFromConfigMap to dataresidency_defaults_test.go

* add store and singleton for data residency

* Add data residency to topic reconciler which is shared by all sources, tested by manually created a Topic and check the created topic in GCP

* Add info logging when updating topic config in sources

* Add data residency support to broker and trigger, not using injection

* Move the common AllowedRegionPolicy computation to dataresidency/default.go

* Fix unit test failure for broker and trigger

* Add more test to TestComputeAllowedPersistenceRegions

* Add singleton test for data residency to work around the covery problem, rename config-data-residency.yaml to make it consistent with other files

* Resolve comments from PR

* Add test to broker to test data residency configuration and labels

* DataResidency test added to trigger

* Add empty data residency config

* Add helper function for data residency configuration map creation

cmd/controller/wire.go

@@ -18,6 +18,7 @@ package main
 import (
 	"context"
 
+	"github.com/google/knative-gcp/pkg/apis/configs/dataresidency"
 	"github.com/google/knative-gcp/pkg/apis/configs/gcpauth"
 	"github.com/google/knative-gcp/pkg/reconciler/events/auditlogs"
 	"github.com/google/knative-gcp/pkg/reconciler/events/build"
@@ -39,6 +40,7 @@ func InitializeControllers(ctx context.Context) ([]injection.ControllerConstruct
 		ClientOptions,
 		iam.PolicyManagerSet,
 		wire.Struct(new(gcpauth.StoreSingleton)),
+		wire.Struct(new(dataresidency.StoreSingleton)),
 		auditlogs.NewConstructor,
 		storage.NewConstructor,
 		scheduler.NewConstructor,

config/config-data-residency.yaml

@@ -0,0 +1 @@
+core/configmaps/data-residency.yaml

config/core/configmaps/data-residency.yaml

@@ -0,0 +1,56 @@
+# Copyright 2020 The Knative 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-dataresidency
+  namespace: cloud-run-events
+  annotations:
+    knative.dev/example-checksum: "5e76f9d5"
+data:
+  default-dataresidency-config: |
+    clusterDefaults:
+      messagestoragepolicy.allowedpersistenceregions: []
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+
+    # default-dataresidency-config is the configuration for determining the default
+    # data residency to apply to all objects that require data residency.
+    # This is expected to be Channels and Sources and Brokers.
+    #
+    # We only support cluster scoped now
+    default-dataresidency-config: |
+      # clusterDefaults are the defaults to apply to every namespace in the
+      # cluster
+      clusterDefaults:
+        # messagestoragepolicy.allowedpersistenceregions field specifies
+        # all the allowed regions for data residency. The default or an empty value will
+        # mean no data residency requirement.
+        messagestoragepolicy.allowedpersistenceregions:
+          - us-east1
+          - us-west1

hack/update-codegen.sh

@@ -77,6 +77,7 @@ ${GOPATH}/bin/deepcopy-gen \
   --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate/boilerplate.go.txt \
   -i github.com/google/knative-gcp/pkg/apis/configs/gcpauth \
   -i github.com/google/knative-gcp/pkg/apis/configs/broker \
+  -i github.com/google/knative-gcp/pkg/apis/configs/dataresidency \
 
 # TODO(yolocs): generate autoscaling v2beta2 in knative/pkg.
 OUTPUT_PKG="github.com/google/knative-gcp/pkg/client/injection/kube" \

pkg/apis/configs/dataresidency/dataresidency_defaults.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2020 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
+
+    https://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 dataresidency
+
+import (
+	"encoding/json"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/yaml"
+)
+
+const (
+	// configName is the name of config map for the default data residency that
+	// GCP resources should use.
+	configName = "config-dataresidency"
+
+	// defaulterKey is the key in the ConfigMap to get the name of the default
+	// DataResidency setting.
+	defaulterKey = "default-dataresidency-config"
+)
+
+// ConfigMapName returns the name of the configmap to read for default data residency settings.
+func ConfigMapName() string {
+	return configName
+}
+
+// NewDefaultsConfigFromConfigMap creates a Defaults from the supplied configMap.
+func NewDefaultsConfigFromConfigMap(config *corev1.ConfigMap) (*Defaults, error) {
+	return NewDefaultsConfigFromMap(config.Data)
+}
+
+// NewDefaultsConfigFromMap creates a Defaults from the supplied Map.
+func NewDefaultsConfigFromMap(data map[string]string) (*Defaults, error) {
+	nc := &Defaults{}
+
+	// Parse out the data residency configuration.
+	value, present := data[defaulterKey]
+	// Data residency configuration is not required, in which case it will mean
+	// allow all regions, so we simply use an empty one
+	if !present || value == "" {
+		return nc, nil
+	}
+	if err := parseEntry(value, nc); err != nil {
+		return nil, fmt.Errorf("failed to parse the entry: %s", err)
+	}
+	return nc, nil
+}
+
+func parseEntry(entry string, out interface{}) error {
+	j, err := yaml.YAMLToJSON([]byte(entry))
+	if err != nil {
+		return fmt.Errorf("ConfigMap's value could not be converted to JSON: %s : %v", err, entry)
+	}
+	return json.Unmarshal(j, &out)
+}

pkg/apis/configs/dataresidency/dataresidency_defaults_test.go

@@ -0,0 +1,160 @@
+/*
+Copyright 2020 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 dataresidency
+
+import (
+	"testing"
+
+	"cloud.google.com/go/pubsub"
+
+	"github.com/google/go-cmp/cmp"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	. "knative.dev/pkg/configmap/testing"
+	_ "knative.dev/pkg/system/testing"
+)
+
+func TestDefaultsConfigurationFromFile(t *testing.T) {
+	_, example := ConfigMapsFromTestFile(t, configName, defaulterKey)
+	if _, err := NewDefaultsConfigFromConfigMap(example); err != nil {
+		t.Errorf("NewDefaultsConfigFromConfigMap(example) = %v", err)
+	}
+}
+
+func TestNewDefaultsConfigFromConfigMap(t *testing.T) {
+	_, example := ConfigMapsFromTestFile(t, configName, defaulterKey)
+	defaults, err := NewDefaultsConfigFromConfigMap(example)
+	if err != nil {
+		t.Fatalf("NewDefaultsConfigFromConfigMap(example) = %v", err)
+	}
+
+	// Only cluster wide configuration is supported now, but we use the namespace
+	// as the test name and for future extension.
+	testCases := []struct {
+		ns      string
+		regions []string
+	}{
+		{
+			ns:      "cluster-wide",
+			regions: []string{"us-east1", "us-west1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.ns, func(t *testing.T) {
+			if diff := cmp.Diff(tc.regions, defaults.AllowedPersistenceRegions()); diff != "" {
+				t.Errorf("Unexpected value (-want +got): %s", diff)
+			}
+		})
+	}
+}
+
+func TestComputeAllowedPersistenceRegions(t *testing.T) {
+	// Only cluster wide configuration is supported now, but we use the namespace
+	// as the test name and for future extension.
+	testCases := []struct {
+		ns                 string
+		topicConfigRegions []string
+		dsRegions          []string
+		expectedRegions    []string
+	}{
+		{
+			ns:                 "subset",
+			topicConfigRegions: []string{"us-east1", "us-west1"},
+			dsRegions:          []string{"us-west1"},
+			expectedRegions:    []string{"us-west1"},
+		},
+		{
+			ns:                 "conflict",
+			topicConfigRegions: []string{"us-east1"},
+			dsRegions:          []string{"us-west1"},
+			expectedRegions:    []string{"us-west1"},
+		},
+		{
+			ns:                 "topic-nil",
+			topicConfigRegions: nil,
+			dsRegions:          []string{"us-west1"},
+			expectedRegions:    []string{"us-west1"},
+		},
+		{
+			ns:                 "topic-nil-ds-empty",
+			topicConfigRegions: nil,
+			dsRegions:          []string{},
+			expectedRegions:    nil,
+		},
+		{
+			ns:                 "ds-empty",
+			topicConfigRegions: []string{"us-east1"},
+			dsRegions:          []string{},
+			expectedRegions:    []string{"us-east1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.ns, func(t *testing.T) {
+			defaults, err := NewDefaultsConfigFromMap(map[string]string{})
+			if err != nil {
+				t.Fatalf("NewDefaultsConfigFromConfigMap(empty) = %v", err)
+			}
+			defaults.ClusterDefaults.AllowedPersistenceRegions = tc.dsRegions
+			topicConfig := &pubsub.TopicConfig{}
+			topicConfig.MessageStoragePolicy.AllowedPersistenceRegions = tc.topicConfigRegions
+			defaults.ComputeAllowedPersistenceRegions(topicConfig)
+			if diff := cmp.Diff(tc.expectedRegions, topicConfig.MessageStoragePolicy.AllowedPersistenceRegions); diff != "" {
+				t.Errorf("Unexpected value (-want +got): %s", diff)
+			}
+		})
+	}
+}
+
+func TestNewDefaultsConfigFromConfigMapEmpty(t *testing.T) {
+	testCases := map[string]struct {
+		name   string
+		config *corev1.ConfigMap
+	}{
+		"empty data": {
+			config: &corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "cloud-run-events",
+					Name:      configName,
+				},
+				Data: map[string]string{},
+			},
+		},
+		"missing key": {
+			config: &corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "cloud-run-events",
+					Name:      configName,
+				},
+				Data: map[string]string{
+					"other-keys": "are-present",
+				},
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			_, err := NewDefaultsConfigFromConfigMap(tc.config)
+			if err != nil {
+				t.Errorf("Empty value or no key should pass")
+			}
+		})
+	}
+}

pkg/apis/configs/dataresidency/defaults.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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 dataresidency
+
+import (
+	"cloud.google.com/go/pubsub"
+)
+
+// Defaults includes the default values to be populated by the Webhook.
+type Defaults struct {
+	// ClusterDefaults are the data residency defaults to use for all namepaces
+	ClusterDefaults ScopedDefaults `json:"clusterDefaults,omitempty"`
+}
+
+// ScopedDefaults are the data residency setting defaults.
+type ScopedDefaults struct {
+	// AllowedPersistenceRegions specifies the regions allowed for data
+	// storage. Eg "us-east1". An empty configuration means no data residency
+	// constraints.
+	AllowedPersistenceRegions []string `json:"messagestoragepolicy.allowedpersistenceregions,omitempty"`
+}
+
+// scoped gets the scoped data residency defaults, for now we only have
+// cluster scope.
+func (d *Defaults) scoped() *ScopedDefaults {
+	scopedDefaults := &d.ClusterDefaults
+	// currently we don't support namespace, but if we do, we should check
+	// namespace default here.
+	return scopedDefaults
+}
+
+// AllowedPersistenceRegions gets the AllowedPersistenceRegions setting in the default.
+func (d *Defaults) AllowedPersistenceRegions() []string {
+	return d.scoped().AllowedPersistenceRegions
+}
+
+// ComputeAllowedPersistenceRegions computes the final message storage policy in
+// topicConfig. Return true if the topicConfig is updated.
+func (d *Defaults) ComputeAllowedPersistenceRegions(topicConfig *pubsub.TopicConfig) bool {
+	// We can do subset of both in the future, but for now, we just overwrite the
+	// configuration as the relationship between region and zones are not clear to handle,
+	// eg. us-east1 vs us-east1-a. Important note: setting the AllowedPersistenceRegions
+	// to empty string slice is an error, should set it to nil for all regions.
+	allowedRegions := d.AllowedPersistenceRegions()
+	if allowedRegions == nil || len(allowedRegions) == 0 {
+		return false
+	}
+
+	topicConfig.MessageStoragePolicy.AllowedPersistenceRegions = allowedRegions
+	return true
+}