<commit_before>/*
Copyright 2020 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 gcs

import (
	"os"
	"path/filepath"
	"strings"

	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"k8s.io/release/pkg/gcp"
	"k8s.io/utils/pointer"
)

var (
	// GcsPrefix url prefix for google cloud storage buckets
	GcsPrefix      = "gs://"
	concurrentFlag = "-m"
	recursiveFlag  = "-r"
	noClobberFlag  = "-n"
)

type Options struct {
	// gsutil options
	Concurrent *bool
	Recursive  *bool
	NoClobber  *bool

	// local options
	// AllowMissing allows a copy operation to be skipped if the source or
	// destination does not exist. This is useful for scenarios where copy
	// operations happen in a loop/channel, so a single "failure" does not block
	// the entire operation.
	AllowMissing *bool
}

// DefaultGCSCopyOptions have the default options for the GCS copy action
var DefaultGCSCopyOptions = &Options{
	Concurrent:   pointer.BoolPtr(true),
	Recursive:    pointer.BoolPtr(true),
	NoClobber:    pointer.BoolPtr(true),
	AllowMissing: pointer.BoolPtr(true),
}

// CopyToGCS copies a local directory to the specified GCS path
// TODO: Consider using isPathNormalized here
func CopyToGCS(src, gcsPath string, opts *Options) error {
	logrus.Infof("Copying %s to GCS (%s)", src, gcsPath)
	gcsPath = NormalizeGCSPath(gcsPath)

	_, err := os.Stat(src)
	if err != nil {
		logrus.Info("Unable to get local source directory info")

		if *opts.AllowMissing {
			logrus.Infof("Source directory (%s) does not exist. Skipping GCS upload.", src)
			return nil
		}

		return errors.New("source directory does not exist")
	}

	return bucketCopy(src, gcsPath, opts)
}

// CopyToLocal copies a GCS path to the specified local directory
// TODO: Consider using isPathNormalized here
func CopyToLocal(gcsPath, dst string, opts *Options) error {
	logrus.Infof("Copying GCS (%s) to %s", gcsPath, dst)
	gcsPath = NormalizeGCSPath(gcsPath)

	return bucketCopy(gcsPath, dst, opts)
}

// CopyBucketToBucket copies between two GCS paths.
// TODO: Consider using isPathNormalized here
func CopyBucketToBucket(src, dst string, opts *Options) error {
	logrus.Infof("Copying %s to %s", src, dst)
	return bucketCopy(NormalizeGCSPath(src), NormalizeGCSPath(dst), opts)
}

func bucketCopy(src, dst string, opts *Options) error {
	args := []string{}

	if *opts.Concurrent {
		logrus.Debug("Setting GCS copy to run concurrently")
		args = append(args, concurrentFlag)
	}

	args = append(args, "cp")
	if *opts.Recursive {
		logrus.Debug("Setting GCS copy to run recursively")
		args = append(args, recursiveFlag)
	}
	if *opts.NoClobber {
		logrus.Debug("Setting GCS copy to not clobber existing files")
		args = append(args, noClobberFlag)
	}

	args = append(args, src, dst)

	if err := gcp.GSUtil(args...); err != nil {
		return errors.Wrap(err, "gcs copy")
	}

	return nil
}

// GetReleasePath returns a GCS path to retrieve builds from or push builds to
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>][/fast][/<version>]
func GetReleasePath(
	bucket, buildType, gcsSuffix, version string,
	fast bool) string {
	gcsPath := getPath(
		bucket,
		buildType,
		gcsSuffix,
		version,
		"release",
		fast,
	)

	logrus.Infof("Release path is %s", gcsPath)
	return gcsPath
}

// GetMarkerPath returns a GCS path where version markers should be stored
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>]
func GetMarkerPath(
	bucket, buildType, gcsSuffix string) string {
	gcsPath := getPath(
		bucket,
		buildType,
		gcsSuffix,
		"",
		"marker",
		false,
	)

	logrus.Infof("Version marker path is %s", gcsPath)
	return gcsPath
}

// GetReleasePath returns a GCS path to retrieve builds from or push builds to
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>][/fast][/<version>]
// TODO: Support "release" buildType
func getPath(
	bucket, buildType, gcsSuffix, version, pathType string,
	fast bool) string {
	gcsPath := bucket
	gcsPath = filepath.Join(gcsPath, buildType)

	if gcsSuffix != "" {
		gcsPath += "-" + gcsSuffix
	}

	if pathType == "release" {
		if fast {
			gcsPath = filepath.Join(gcsPath, "fast")
		}

		if version != "" {
			gcsPath = filepath.Join(gcsPath, version)
		}
	}

	// Ensure any constructed GCS path is prefixed with `gs://`
	gcsPath = NormalizeGCSPath(gcsPath)

	return gcsPath
}

// NormalizeGCSPath takes a GCS path and ensures that the `GcsPrefix` is
// prepended to it.
func NormalizeGCSPath(gcsPath string) string {
	gcsPath = strings.TrimPrefix(gcsPath, GcsPrefix)
	gcsPath = GcsPrefix + gcsPath

	return gcsPath
}

// isPathNormalized determines if a GCS path is prefixed with `gs://`.
// Use this function as pre-check for any gsutil/GCS functions that manipulate
// GCS bucket contents.
func isPathNormalized(gcsPath string) bool {
	if !strings.HasPrefix(gcsPath, GcsPrefix) {
		logrus.Errorf("GCS path (%s) should be prefixed with `gs://`", GcsPrefix)

		return false
	}

	return true
}

// RsyncRecursive runs `gsutil rsync` in recursive mode. The caller of this
// function has to ensure that the provided paths are prefixed with gs:// if
// necessary (see `NormalizeGCSPath()`).
// TODO: Implementation of `gsutil rsync` should support local directory copies
//       with `-d`
func RsyncRecursive(src, dst string) error {<commit_msg>pkg/gcp/gcs: NormalizeGCSPath now handles multiple path elements<commit_after>/*
Copyright 2020 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 gcs

import (
	"os"
	"path/filepath"
	"strings"

	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"k8s.io/release/pkg/gcp"
	"k8s.io/utils/pointer"
)

var (
	// GcsPrefix url prefix for google cloud storage buckets
	GcsPrefix      = "gs://"
	concurrentFlag = "-m"
	recursiveFlag  = "-r"
	noClobberFlag  = "-n"
)

type Options struct {
	// gsutil options
	Concurrent *bool
	Recursive  *bool
	NoClobber  *bool

	// local options
	// AllowMissing allows a copy operation to be skipped if the source or
	// destination does not exist. This is useful for scenarios where copy
	// operations happen in a loop/channel, so a single "failure" does not block
	// the entire operation.
	AllowMissing *bool
}

// DefaultGCSCopyOptions have the default options for the GCS copy action
var DefaultGCSCopyOptions = &Options{
	Concurrent:   pointer.BoolPtr(true),
	Recursive:    pointer.BoolPtr(true),
	NoClobber:    pointer.BoolPtr(true),
	AllowMissing: pointer.BoolPtr(true),
}

// CopyToGCS copies a local directory to the specified GCS path
// TODO: Consider using isPathNormalized here
func CopyToGCS(src, gcsPath string, opts *Options) error {
	logrus.Infof("Copying %s to GCS (%s)", src, gcsPath)
	gcsPath = NormalizeGCSPath(gcsPath)

	_, err := os.Stat(src)
	if err != nil {
		logrus.Info("Unable to get local source directory info")

		if *opts.AllowMissing {
			logrus.Infof("Source directory (%s) does not exist. Skipping GCS upload.", src)
			return nil
		}

		return errors.New("source directory does not exist")
	}

	return bucketCopy(src, gcsPath, opts)
}

// CopyToLocal copies a GCS path to the specified local directory
// TODO: Consider using isPathNormalized here
func CopyToLocal(gcsPath, dst string, opts *Options) error {
	logrus.Infof("Copying GCS (%s) to %s", gcsPath, dst)
	gcsPath = NormalizeGCSPath(gcsPath)

	return bucketCopy(gcsPath, dst, opts)
}

// CopyBucketToBucket copies between two GCS paths.
// TODO: Consider using isPathNormalized here
func CopyBucketToBucket(src, dst string, opts *Options) error {
	logrus.Infof("Copying %s to %s", src, dst)
	return bucketCopy(NormalizeGCSPath(src), NormalizeGCSPath(dst), opts)
}

func bucketCopy(src, dst string, opts *Options) error {
	args := []string{}

	if *opts.Concurrent {
		logrus.Debug("Setting GCS copy to run concurrently")
		args = append(args, concurrentFlag)
	}

	args = append(args, "cp")
	if *opts.Recursive {
		logrus.Debug("Setting GCS copy to run recursively")
		args = append(args, recursiveFlag)
	}
	if *opts.NoClobber {
		logrus.Debug("Setting GCS copy to not clobber existing files")
		args = append(args, noClobberFlag)
	}

	args = append(args, src, dst)

	if err := gcp.GSUtil(args...); err != nil {
		return errors.Wrap(err, "gcs copy")
	}

	return nil
}

// GetReleasePath returns a GCS path to retrieve builds from or push builds to
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>][/fast][/<version>]
func GetReleasePath(
	bucket, buildType, gcsSuffix, version string,
	fast bool) string {
	gcsPath := getPath(
		bucket,
		buildType,
		gcsSuffix,
		version,
		"release",
		fast,
	)

	logrus.Infof("Release path is %s", gcsPath)
	return gcsPath
}

// GetMarkerPath returns a GCS path where version markers should be stored
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>]
func GetMarkerPath(
	bucket, buildType, gcsSuffix string) string {
	gcsPath := getPath(
		bucket,
		buildType,
		gcsSuffix,
		"",
		"marker",
		false,
	)

	logrus.Infof("Version marker path is %s", gcsPath)
	return gcsPath
}

// GetReleasePath returns a GCS path to retrieve builds from or push builds to
//
// Expected destination format:
//   gs://<bucket>/<buildType>[-<gcsSuffix>][/fast][/<version>]
// TODO: Support "release" buildType
func getPath(
	bucket, buildType, gcsSuffix, version, pathType string,
	fast bool) string {
	gcsPath := bucket
	gcsPath = filepath.Join(gcsPath, buildType)

	if gcsSuffix != "" {
		gcsPath += "-" + gcsSuffix
	}

	if pathType == "release" {
		if fast {
			gcsPath = filepath.Join(gcsPath, "fast")
		}

		if version != "" {
			gcsPath = filepath.Join(gcsPath, version)
		}
	}

	// Ensure any constructed GCS path is prefixed with `gs://`
	gcsPath = NormalizeGCSPath(gcsPath)

	return gcsPath
}

// NormalizeGCSPath takes a GCS path and ensures that the `GcsPrefix` is
// prepended to it.
func NormalizeGCSPath(gcsPathParts ...string) (string, error) {
	gcsPath := ""

	// Ensure there is at least one element in the gcsPathParts slice before
	// trying to construct a path
	if len(gcsPathParts) == 0 {
		return "", errors.New("must contain at least one path part")
	} else if len(gcsPathParts) == 1 {
		if gcsPathParts[0] == "" {
			return "", errors.New("path should not be an empty string")
		}

		gcsPath = gcsPathParts[0]
	} else {
		gcsPath = filepath.Join(gcsPathParts...)
	}

	// Strip `gs://` if it was included in gcsPathParts
	gcsPath = strings.TrimPrefix(gcsPath, GcsPrefix)

	// Strip `gs:/` if:
	// - `gs://` was included in gcsPathParts
	// - gcsPathParts had more than element
	// - filepath.Join() was called somewhere in a caller's logic
	gcsPath = strings.TrimPrefix(gcsPath, "gs:/")

	// Strip `/`
	// This scenario may never happen, but let's catch it, just in case
	gcsPath = strings.TrimPrefix(gcsPath, "/")

	gcsPath = GcsPrefix + gcsPath

	return gcsPath, nil
}

// isPathNormalized determines if a GCS path is prefixed with `gs://`.
// Use this function as pre-check for any gsutil/GCS functions that manipulate
// GCS bucket contents.
func isPathNormalized(gcsPath string) bool {
	if !strings.HasPrefix(gcsPath, GcsPrefix) {
		logrus.Errorf("GCS path (%s) should be prefixed with `gs://`", GcsPrefix)

		return false
	}

	return true
}

// RsyncRecursive runs `gsutil rsync` in recursive mode. The caller of this
// function has to ensure that the provided paths are prefixed with gs:// if
// necessary (see `NormalizeGCSPath()`).
// TODO: Implementation of `gsutil rsync` should support local directory copies
//       with `-d`
func RsyncRecursive(src, dst string) error {
{'fraction_non_alphanumeric': 0.11145634161886551, 'fraction_numerical': 0.0019120458891013384, 'mean_word_length': 3.9304791830322072, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3992652', 'n_tokens_mistral': 4332, 'n_tokens_neox': 4128, 'n_words': 1377}