Merge pull request #1305 from sunnylovestiramisu/automated-cherry-pick-of-#1101-#1227-#1296-upstream-release-1.7

Automated cherry pick of #1101: Add provisionedThroughput for hyperdisk
#1227: Adding new metric pdcsi_operation_errors to fetch error
#1296: emit metrics even for success scenarios

Author: k8s-ci-robot

README.md

@@ -65,7 +65,7 @@ See Github [Issues](https://github.com/kubernetes-sigs/gcp-compute-persistent-di
 | disk-encryption-kms-key     | Fully qualified resource identifier for the key to use to encrypt new disks. | Empty string. | Encrypt disk using Customer Managed Encryption Key (CMEK). See [GKE Docs](https://cloud.google.com/kubernetes-engine/docs/how-to/using-cmek#create_a_cmek_protected_attached_disk) for details. |
 | labels                      | `key1=value1,key2=value2` |               | Labels allow you to assign custom [GCE Disk labels](https://cloud.google.com/compute/docs/labeling-resources). |
 | provisioned-iops-on-create  | string (int64 format). Values typically between 10,000 and 120,000 |               | Indicates how many IOPS to provision for the disk. See the [Extreme persistent disk documentation](https://cloud.google.com/compute/docs/disks/extreme-persistent-disk) for details, including valid ranges for IOPS. |
-
+| provisioned-throughput-on-create  | string (int64 format). Values typically between 1 and 7,124 mb per second |               | Indicates how much throughput to provision for the disk. See the [hyperdisk documentation](TBD) for details, including valid ranges for throughput. |
 
 ### Topology
 

cmd/gce-pd-csi-driver/main.go

@@ -91,10 +91,14 @@ func handle() {
 	}
 	klog.V(2).Infof("Driver vendor version %v", version)
 
-	if *runControllerService && *httpEndpoint != "" && metrics.IsGKEComponentVersionAvailable() {
+	if *runControllerService && *httpEndpoint != "" {
 		mm := metrics.NewMetricsManager()
 		mm.InitializeHttpHandler(*httpEndpoint, *metricsPath)
-		mm.EmitGKEComponentVersion()
+		mm.RegisterPDCSIMetric()
+
+		if metrics.IsGKEComponentVersionAvailable() {
+			mm.EmitGKEComponentVersion()
+		}
 	}
 
 	if len(*extraVolumeLabelsStr) > 0 && !*runControllerService {

pkg/common/parameters.go

@@ -23,11 +23,12 @@ import (
 
 const (
 	// Parameters for StorageClass
-	ParameterKeyType                    = "type"
-	ParameterKeyReplicationType         = "replication-type"
-	ParameterKeyDiskEncryptionKmsKey    = "disk-encryption-kms-key"
-	ParameterKeyLabels                  = "labels"
-	ParameterKeyProvisionedIOPSOnCreate = "provisioned-iops-on-create"
+	ParameterKeyType                          = "type"
+	ParameterKeyReplicationType               = "replication-type"
+	ParameterKeyDiskEncryptionKmsKey          = "disk-encryption-kms-key"
+	ParameterKeyLabels                        = "labels"
+	ParameterKeyProvisionedIOPSOnCreate       = "provisioned-iops-on-create"
+	ParameterKeyProvisionedThroughputOnCreate = "provisioned-throughput-on-create"
 
 	// Parameters for VolumeSnapshotClass
 	ParameterKeyStorageLocations = "storage-locations"
@@ -79,6 +80,9 @@ type DiskParameters struct {
 	// Values: {int64}
 	// Default: none
 	ProvisionedIOPSOnCreate int64
+	// Values: {int64}
+	// Default: none
+	ProvisionedThroughputOnCreate int64
 }
 
 // SnapshotParameters contains normalized and defaulted parameters for snapshots
@@ -144,6 +148,12 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 				return p, fmt.Errorf("parameters contain invalid provisionedIOPSOnCreate parameter: %w", err)
 			}
 			p.ProvisionedIOPSOnCreate = paramProvisionedIOPSOnCreate
+		case ParameterKeyProvisionedThroughputOnCreate:
+			paramProvisionedThroughputOnCreate, err := ConvertMiBStringToInt64(v)
+			if err != nil {
+				return p, fmt.Errorf("parameters contain invalid provisionedThroughputOnCreate parameter: %w", err)
+			}
+			p.ProvisionedThroughputOnCreate = paramProvisionedThroughputOnCreate
 		default:
 			return p, fmt.Errorf("parameters contains invalid option %q", k)
 		}

pkg/common/parameters_test.go

@@ -90,6 +90,22 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 				ProvisionedIOPSOnCreate: 10000,
 			},
 		},
+		{
+			name:       "values from parameters, checking hyperdisk-throughput",
+			parameters: map[string]string{ParameterKeyType: "hyperdisk-throughput", ParameterKeyReplicationType: "none", ParameterKeyDiskEncryptionKmsKey: "foo/key", ParameterKeyLabels: "key1=value1,key2=value2", ParameterKeyProvisionedThroughputOnCreate: "1000Mi"},
+			labels:     map[string]string{},
+			expectParams: DiskParameters{
+				DiskType:             "hyperdisk-throughput",
+				ReplicationType:      "none",
+				DiskEncryptionKMSKey: "foo/key",
+				Tags:                 map[string]string{},
+				Labels: map[string]string{
+					"key1": "value1",
+					"key2": "value2",
+				},
+				ProvisionedThroughputOnCreate: 1000,
+			},
+		},
 		{
 			name:       "values from parameters, checking balanced pd",
 			parameters: map[string]string{ParameterKeyType: "pd-balanced", ParameterKeyReplicationType: "regional-pd", ParameterKeyDiskEncryptionKmsKey: "foo/key"},

pkg/common/utils.go

@@ -17,14 +17,21 @@ limitations under the License.
 package common
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"regexp"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
+	"google.golang.org/api/googleapi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/sets"
 	volumehelpers "k8s.io/cloud-provider/volume/helpers"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -279,3 +286,94 @@ func ConvertStringToInt64(str string) (int64, error) {
 	}
 	return volumehelpers.RoundUpToB(quantity)
 }
+
+// ConvertMiBStringToInt64 converts a GiB string to int64
+func ConvertMiBStringToInt64(str string) (int64, error) {
+	// Verify regex before
+	match, _ := regexp.MatchString("^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$", str)
+	if !match {
+		return 0, fmt.Errorf("invalid string %s", str)
+	}
+	quantity := resource.MustParse(str)
+	return volumehelpers.RoundUpToMiB(quantity)
+}
+
+// CodeForError returns a pointer to the grpc error code that maps to the http
+// error code for the passed in user googleapi error or context error. Returns
+// codes.Internal if the given error is not a googleapi error caused by the user.
+// The following http error codes are considered user errors:
+// (1) http 400 Bad Request, returns grpc InvalidArgument,
+// (2) http 403 Forbidden, returns grpc PermissionDenied,
+// (3) http 404 Not Found, returns grpc NotFound
+// (4) http 429 Too Many Requests, returns grpc ResourceExhausted
+// The following errors are considered context errors:
+// (1) "context deadline exceeded", returns grpc DeadlineExceeded,
+// (2) "context canceled", returns grpc Canceled
+func CodeForError(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+
+	if errCode := existingErrorCode(err); errCode != nil {
+		return errCode
+	}
+	if code := isContextError(err); code != nil {
+		return code
+	}
+
+	internalErrorCode := codes.Internal
+	// Upwrap the error
+	var apiErr *googleapi.Error
+	if !errors.As(err, &apiErr) {
+		return &internalErrorCode
+	}
+
+	userErrors := map[int]codes.Code{
+		http.StatusForbidden:       codes.PermissionDenied,
+		http.StatusBadRequest:      codes.InvalidArgument,
+		http.StatusTooManyRequests: codes.ResourceExhausted,
+		http.StatusNotFound:        codes.NotFound,
+	}
+	if code, ok := userErrors[apiErr.Code]; ok {
+		return &code
+	}
+
+	return &internalErrorCode
+}
+
+// isContextError returns a pointer to the grpc error code DeadlineExceeded
+// if the passed in error contains the "context deadline exceeded" string and returns
+// the grpc error code Canceled if the error contains the "context canceled" string.
+func isContextError(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+
+	errStr := err.Error()
+	if strings.Contains(errStr, context.DeadlineExceeded.Error()) {
+		return errCodePtr(codes.DeadlineExceeded)
+	}
+	if strings.Contains(errStr, context.Canceled.Error()) {
+		return errCodePtr(codes.Canceled)
+	}
+	return nil
+}
+
+func existingErrorCode(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+	if status, ok := status.FromError(err); ok {
+		return errCodePtr(status.Code())
+	}
+	return nil
+}
+
+func errCodePtr(code codes.Code) *codes.Code {
+	return &code
+}
+
+func LoggedError(msg string, err error) error {
+	klog.Errorf(msg+"%v", err.Error())
+	return status.Errorf(*CodeForError(err), msg+"%v", err.Error())
+}