/*
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 deployer

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	"github.com/blang/semver/v4"
	"k8s.io/klog/v2"
	"k8s.io/kops/tests/e2e/kubetest2-kops/aws"
	"k8s.io/kops/tests/e2e/kubetest2-kops/gce"
	"k8s.io/kops/tests/e2e/pkg/target"
	"k8s.io/kops/tests/e2e/pkg/util"
	"sigs.k8s.io/kubetest2/pkg/boskos"
)

func (d *deployer) init() error {
	var err error
	d.doInit.Do(func() { err = d.initialize() })
	return err
}

// initialize should only be called by init(), behind a sync.Once
func (d *deployer) initialize() error {
	if d.commonOptions.ShouldUp() || d.commonOptions.ShouldDown() {
		if err := d.verifyKopsFlags(); err != nil {
			return fmt.Errorf("init failed to check kops flags: %v", err)
		}
	}
	if d.commonOptions.ShouldUp() {
		if err := d.verifyUpFlags(); err != nil {
			return fmt.Errorf("init failed to check up flags: %v", err)
		}
	}

	var err error
	d.zones, err = d.getZones()
	if err != nil {
		return err
	}
	switch d.CloudProvider {
	case "aws":
		if d.region == "" {
			// Default to us-east-2, but use the region implied by the first zone if possible
			if len(d.zones) == 0 {
				d.region = "us-east-2"
			} else {
				d.region = d.zones[0][:len(d.zones[0])-1]
			}
		}
		client, err := aws.NewClient(context.Background(), d.region)
		if err != nil {
			return fmt.Errorf("init failed to build AWS client: %w", err)
		}
		d.aws = client

		if d.SSHPrivateKeyPath == "" {
			d.SSHPrivateKeyPath = os.Getenv("AWS_SSH_PRIVATE_KEY_FILE")
		}
		if d.SSHPublicKeyPath == "" {
			d.SSHPublicKeyPath = os.Getenv("AWS_SSH_PUBLIC_KEY_FILE")
		}
		if d.SSHPrivateKeyPath == "" || d.SSHPublicKeyPath == "" {
			publicKeyPath, privateKeyPath, err := util.CreateSSHKeyPair(d.ClusterName)
			if err != nil {
				return err
			}
			d.SSHPublicKeyPath = publicKeyPath
			d.SSHPrivateKeyPath = privateKeyPath
		}
	case "azure":
		publicKeyPath, privateKeyPath, err := util.CreateSSHKeyPair(d.ClusterName)
		if err != nil {
			return err
		}
		d.SSHPublicKeyPath = publicKeyPath
		d.SSHPrivateKeyPath = privateKeyPath
		d.SSHUser = "kops"
	case "digitalocean":
		if d.SSHPrivateKeyPath == "" {
			d.SSHPrivateKeyPath = os.Getenv("DO_SSH_PRIVATE_KEY_FILE")
		}
		if d.SSHPublicKeyPath == "" {
			d.SSHPublicKeyPath = os.Getenv("DO_SSH_PUBLIC_KEY_FILE")
		}
		d.SSHUser = "root"
	case "gce":
		d.region, err = gce.ZoneToRegion(d.zones[0])
		if err != nil {
			return err
		}
		if d.GCPProject == "" {
			klog.V(1).Info("No GCP project provided, acquiring from Boskos")

			boskosClient, err := boskos.NewClient(d.BoskosLocation)
			if err != nil {
				return fmt.Errorf("failed to make boskos client: %s", err)
			}
			d.boskos = boskosClient

			resource, err := boskos.Acquire(
				d.boskos,
				d.BoskosResourceType,
				d.BoskosAcquireTimeout,
				d.BoskosHeartbeatInterval,
				d.boskosHeartbeatClose,
			)
			if err != nil {
				return fmt.Errorf("init failed to get project from boskos: %s", err)
			}
			d.GCPProject = resource.Name
			klog.V(1).Infof("Got project %s from boskos", d.GCPProject)

			if d.SSHPrivateKeyPath == "" {
				d.SSHPrivateKeyPath = os.Getenv("GCE_SSH_PRIVATE_KEY_FILE")
			}
			if d.SSHPublicKeyPath == "" {
				d.SSHPublicKeyPath = os.Getenv("GCE_SSH_PUBLIC_KEY_FILE")
			}
			if d.SSHPrivateKeyPath == "" && d.SSHPublicKeyPath == "" {
				privateKey, publicKey, err := gce.SetupSSH(d.GCPProject)
				if err != nil {
					return err
				}
				d.SSHPrivateKeyPath = privateKey
				d.SSHPublicKeyPath = publicKey
			}

		} else if d.SSHPrivateKeyPath == "" && os.Getenv("KUBE_SSH_KEY_PATH") != "" {
			d.SSHPrivateKeyPath = os.Getenv("KUBE_SSH_KEY_PATH")
		}
	}

	klog.V(1).Infof("Using SSH keypair: [%s,%s]", d.SSHPrivateKeyPath, d.SSHPublicKeyPath)

	// Determine whether ephemeral buckets need to be created. Each store
	// method generates a dynamic bucket name when its corresponding env var
	// is unset; those buckets must be created before cluster provisioning
	// and deleted during teardown.
	switch d.CloudProvider {
	case "aws":
		if os.Getenv("KOPS_STATE_STORE") == "" {
			d.createStateStore = true
		}
		if _, found := os.LookupEnv("KOPS_DISCOVERY_STORE"); !found {
			d.createDiscoveryStore = true
		}
	case "gce":
		if d.boskos != nil || os.Getenv("KOPS_STATE_STORE") == "" || os.Getenv("KOPS_STAGING_BUCKET") == "" {
			d.createStateStore = true
		}
	}

	if d.commonOptions.ShouldBuild() {
		if err := d.verifyBuildFlags(); err != nil {
			return fmt.Errorf("init failed to check build flags: %v", err)
		}
	}

	if d.SSHUser == "" {
		d.SSHUser = os.Getenv("KUBE_SSH_USER")
	}
	klog.V(1).Infof("Using SSH user: [%s]", d.SSHUser)

	if d.TerraformVersion != "" {
		t, err := target.NewTerraform(d.TerraformVersion, d.ArtifactsDir)
		if err != nil {
			return err
		}
		d.terraform = t
	}
	if d.commonOptions.ShouldTest() {
		for _, envvar := range d.env() {
			// Set all of the env vars we use for kops in the current process
			// so that the tester inherits them when shelling out to kops
			if i := strings.Index(envvar, "="); i != -1 {
				os.Setenv(envvar[0:i], envvar[i+1:])
			} else {
				os.Setenv(envvar, "")
			}
		}
	}
	return nil
}

// verifyKopsFlags ensures common fields are set for kops commands
func (d *deployer) verifyKopsFlags() error {
	if d.ClusterName == "" {
		name, err := d.defaultClusterName()
		if err != nil {
			return err
		}
		d.ClusterName = name
		klog.Infof("Using cluster name: %v", d.ClusterName)
	}

	if d.KopsBinaryPath == "" && d.KopsVersionMarker == "" && d.KopsVersion == "" {
		return errors.New("atleast one of  --kops-binary-path, --kops-version-marker, --kops-version must be set")
	}
	if d.KopsVersionMarker != "" && d.KopsVersion != "" {
		return errors.New("you can't set kops-version-marker and kops-version at the same time")
	}
	if d.KopsBinaryPath != "" && (d.KopsVersion != "" || d.KopsVersionMarker != "") {
		return errors.New("you can't set kops-binary-path with kops-version-marker or kops-version at the same time")
	}

	if d.ControlPlaneCount == 0 {
		d.ControlPlaneCount = 1
	}

	switch d.CloudProvider {
	case "aws":
	case "azure":
	case "gce":
	case "digitalocean":
	default:
		return errors.New("unsupported --cloud-provider value")
	}

	return nil
}

// env returns a list of environment variables passed to the kops binary
func (d *deployer) env() []string {
	vars := d.Env
	vars = append(vars, []string{
		fmt.Sprintf("PATH=%v", os.Getenv("PATH")),
		fmt.Sprintf("HOME=%v", os.Getenv("HOME")),
		fmt.Sprintf("KOPS_STATE_STORE=%v", d.stateStore()),
		fmt.Sprintf("KOPS_FEATURE_FLAGS=%v", d.featureFlags()),
		"KOPS_RUN_TOO_NEW_VERSION=1",
	}...)

	if d.BuildOptions.TargetBuildArch != "" {
		vars = append(vars, fmt.Sprintf("KOPS_ARCH=%s", strings.Trim(d.BuildOptions.TargetBuildArch, "linux/")))
	}

	// Pass-through some env vars if set (on all clouds)
	for _, k := range []string{"KOPS_ARCH"} {
		if v := os.Getenv(k); v != "" {
			vars = append(vars, k+"="+v)
		}
	}

	switch d.CloudProvider {
	case "aws":
		// Pass through some env vars if set
		for _, k := range []string{"AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE", "AWS_CONTAINER_CREDENTIALS_FULL_URI", "AWS_PROFILE", "AWS_SHARED_CREDENTIALS_FILE"} {
			v := os.Getenv(k)
			if v != "" {
				vars = append(vars, k+"="+v)
			}
		}
		// Recognized by the e2e framework
		// https://github.com/kubernetes/kubernetes/blob/a750d8054a6cb3167f495829ce3e77ab0ccca48e/test/e2e/framework/ssh/ssh.go#L59-L62
		vars = append(vars, fmt.Sprintf("KUBE_SSH_KEY_PATH=%v", d.SSHPrivateKeyPath))
	case "azure":
		// Pass through some env vars if set
		for _, k := range []string{"AZURE_TENANT_ID", "AZURE_SUBSCRIPTION_ID", "AZURE_CLIENT_ID", "AZURE_FEDERATED_TOKEN_FILE"} {
			v := os.Getenv(k)
			if v != "" {
				vars = append(vars, k+"="+v)
			} else {
				klog.Warningf("Azure env var %q not found or empty", k)
			}
		}
	case "digitalocean":
		// Pass through some env vars if set
		for _, k := range []string{"DIGITALOCEAN_ACCESS_TOKEN", "S3_ACCESS_KEY_ID", "S3_SECRET_ACCESS_KEY"} {
			v := os.Getenv(k)
			if v != "" {
				vars = append(vars, k+"="+v)
			} else {
				klog.Warningf("DO env var %q not found or empty", k)
			}
		}
	case "gce":
		if d.GCPProject != "" {
			vars = append(vars, fmt.Sprintf("GCP_PROJECT=%v", d.GCPProject))
			// ClusterLoader2's managed Prometheus client reads the GCP project
			// from the PROJECT env var, so also expose it under that name.
			vars = append(vars, fmt.Sprintf("PROJECT=%v", d.GCPProject))
		}
	}

	if d.KopsBaseURL != "" {
		vars = append(vars, fmt.Sprintf("KOPS_BASE_URL=%v", d.maybeGSURL(d.KopsBaseURL)))
	} else if baseURL := os.Getenv("KOPS_BASE_URL"); baseURL != "" {
		vars = append(vars, fmt.Sprintf("KOPS_BASE_URL=%v", d.maybeGSURL(baseURL)))
	}

	if kopsBin := d.resolvedKopsBinaryPath(); kopsBin != "" {
		vars = append(vars, fmt.Sprintf("KOPS=%v", kopsBin))
	}

	// Pass through OpenTelemetry flags
	{
		foundOTEL := false
		for _, k := range []string{
			"OTEL_EXPORTER_OTLP_TRACES_FILE", "OTEL_EXPORTER_OTLP_FILE",
			"OTEL_EXPORTER_OTLP_TRACES_DIR", "OTEL_EXPORTER_OTLP_DIR",
		} {
			v := os.Getenv(k)
			if v != "" {
				foundOTEL = true
				vars = append(vars, k+"="+v)
			}
		}

		// If no otel flags were explicitly specified, and we have artifacts, log under the artifacts directory
		if !foundOTEL {
			artifacts := d.ArtifactsDir
			if artifacts != "" {
				vars = append(vars, "OTEL_EXPORTER_OTLP_TRACES_DIR="+filepath.Join(artifacts, "otlp"))
			}
		}
	}
	return vars
}

// gcsPublicPrefix is the https form of a GCS bucket, as used for the staged build artifacts.
const gcsPublicPrefix = "https://storage.googleapis.com/"

// maybeGSURL converts baseURL from the public GCS https form to the gs:// form on GCE, so that
// nodes download the staged artifacts with their instance service-account credentials. Any other
// URL, and any other cloud provider, passes through unchanged. Only kops invocations get the gs://
// form; the scripts that download the kops binary run outside the deployer and keep using https.
func (d *deployer) maybeGSURL(baseURL string) string {
	if d.CloudProvider != "gce" || !strings.HasPrefix(baseURL, gcsPublicPrefix) {
		return baseURL
	}
	// Only kops >= 1.37.0-alpha.1 supports a gs:// base URL. The deployer also tests release
	// branches, which must keep the https form. The staged artifacts path always ends in the
	// kops version; if none parses, keep https, which every version can download.
	segments := strings.Split(strings.TrimRight(baseURL, "/"), "/")
	version, err := semver.ParseTolerant(segments[len(segments)-1])
	if err != nil || version.LT(semver.MustParse("1.37.0-alpha.1")) {
		return baseURL
	}
	return "gs://" + strings.TrimPrefix(baseURL, gcsPublicPrefix)
}

// featureFlags returns the kops feature flags to set
func (d *deployer) featureFlags() string {
	// The sflags library splits comma-separated values into separate slice
	// elements, so --env=KOPS_FEATURE_FLAGS=A,B,C becomes ["KOPS_FEATURE_FLAGS=A", "B", "C"].
	// We need to reassemble the original value by collecting all entries from the
	// KOPS_FEATURE_FLAGS entry until we hit another NAME=VALUE pattern.
	var parts []string
	collecting := false
	for _, env := range d.Env {
		if value, ok := strings.CutPrefix(env, "KOPS_FEATURE_FLAGS="); ok {
			parts = append(parts, value)
			collecting = true
		} else if collecting {
			if strings.Contains(env, "=") {
				// Hit another env var, stop collecting
				break
			}
			parts = append(parts, env)
		}
	}
	if len(parts) > 0 {
		return strings.Join(parts, ",")
	}
	// if not set by the env flag, but set in the environment, use that.
	if e := os.Getenv("KOPS_FEATURE_FLAGS"); e != "" {
		return e
	}
	return ""
}

// defaultClusterName returns a kops cluster name to use when ClusterName is not set
func (d *deployer) defaultClusterName() (string, error) {
	dnsDomain := os.Getenv("KOPS_DNS_DOMAIN")
	jobName := os.Getenv("JOB_NAME")
	jobType := os.Getenv("JOB_TYPE")
	buildID := os.Getenv("BUILD_ID")
	pullNumber := os.Getenv("PULL_NUMBER")
	if dnsDomain == "" {
		dnsDomain = "tests-kops-aws.k8s.io"
	}
	if jobName == "" || buildID == "" {
		return "", errors.New("JOB_NAME, and BUILD_ID env vars are required when --cluster-name is not set")
	}
	if jobType == "presubmit" && pullNumber == "" {
		return "", errors.New("PULL_NUMBER must be set when JOB_TYPE=presubmit and --cluster-name is not set")
	}

	// TODO(hakman): revisit how dnsDomain is used, ignoring it is unexpected.
	var suffix string
	switch d.CloudProvider {
	case "aws":
		if strings.Contains(d.CreateArgs, "--dns=none") {
			suffix = "k8s.local"
		} else {
			suffix = dnsDomain
		}
	case "azure":
		if dnsDomain == "k8s.local" || strings.HasSuffix(dnsDomain, ".k8s.local") {
			// Use the k8s.local KOPS_DNS_DOMAIN as the cluster name suffix.
			suffix = dnsDomain
		} else {
			// With --dns=none and the domain is not needed.
			suffix = ""
		}
	default:
		suffix = "k8s.local"
	}

	// Most pull request jobs have the "pull-" prefix.
	jobName = strings.TrimPrefix(jobName, "pull-")

	// SNS has char limit of 80
	if len(jobName) > 79 {
		jobName = jobName[:79]
	}

	if jobType == "presubmit" {
		jobName = fmt.Sprintf("pr%s-%s", pullNumber, jobName)
	} else {
		jobName = fmt.Sprintf("e2e-%s", jobName)
	}

	// GCP has char limit of 64
	gcpLimit := 63 - (len(suffix) + 1) // 1 for the dot
	if len(jobName) > gcpLimit && d.CloudProvider == "gce" {
		jobName = jobName[:gcpLimit]
		jobName = strings.TrimRight(jobName, "-")
	}

	if suffix != "" {
		jobName = jobName + "." + suffix
	}

	return jobName, nil
}

// stateStore returns the kops state store to use
// defaulting to values used in prow jobs
func (d *deployer) stateStore() string {
	if d.stateStoreName != "" {
		return d.stateStoreName
	}
	ss := os.Getenv("KOPS_STATE_STORE")
	if ss == "" {
		switch d.CloudProvider {
		case "aws":
			ctx := context.Background()
			bucketName, err := d.aws.BucketName(ctx, aws.BucketTypeStateStore)
			if err != nil {
				klog.Fatalf("Failed to generate bucket name: %v", err)
				return ""
			}
			ss = "s3://" + bucketName
		case "azure":
			// TODO: Use dynamic container name
			ss = "azureblob://stkopsstatestore/cluster-state"
		case "gce":
			ss = "gs://" + gce.GCSBucketName(d.GCPProject, "state")
		case "digitalocean":
			ss = "do://e2e-kops-space"
		}
	}

	d.stateStoreName = ss
	return ss
}

// discoveryStore returns the VFS path to use for public OIDC documents
func (d *deployer) discoveryStore() string {
	if d.discoveryStoreName != "" {
		return d.discoveryStoreName
	}
	discovery, found := os.LookupEnv("KOPS_DISCOVERY_STORE")
	if !found {
		switch d.CloudProvider {
		case "aws":
			ctx := context.Background()
			bucketName, err := d.aws.BucketName(ctx, aws.BucketTypeDiscoveryStore)
			if err != nil {
				klog.Fatalf("Failed to generate bucket name: %v", err)
				return ""
			}
			discovery = "s3://" + bucketName
		}
	}
	d.discoveryStoreName = discovery
	return discovery
}

func (d *deployer) stagingStore() string {
	if d.stagingStoreName != "" {
		return d.stagingStoreName
	}
	sb := os.Getenv("KOPS_STAGING_BUCKET")
	if sb == "" {
		switch d.CloudProvider {
		case "gce":
			sb = "gs://" + gce.GCSBucketName(d.GCPProject, "staging")
		}
	}
	d.stagingStoreName = sb
	return sb
}

// resolvedKopsBinaryPath returns the path where the kops binary either is or will be placed.
// When --kops-binary-path is provided it returns that value directly.
// When --kops-version-marker or --kops-version is used, Up() downloads the binary to a
// deterministic location under RunDir; this method returns that same path so callers
// (including env()) can reference it before Up() has run.
func (d *deployer) resolvedKopsBinaryPath() string {
	if d.KopsBinaryPath != "" {
		return d.KopsBinaryPath
	}
	if d.KopsVersionMarker != "" || d.KopsVersion != "" {
		return filepath.Join(d.commonOptions.RunDir(), "kops")
	}
	return ""
}

// the default is $ARTIFACTS if set, otherwise ./_artifacts
// constructed as an absolute path to help the ginkgo tester because
// for some reason it needs an absolute path to the kubeconfig
func defaultArtifactsDir() (string, error) {
	if path, set := os.LookupEnv("ARTIFACTS"); set {
		absPath, err := filepath.Abs(path)
		if err != nil {
			return "", fmt.Errorf("failed to convert filepath from $ARTIFACTS (%s) to absolute path: %s", path, err)
		}
		return absPath, nil
	}

	absPath, err := filepath.Abs("_artifacts")
	if err != nil {
		return "", fmt.Errorf("when constructing default artifacts dir, failed to get absolute path: %s", err)
	}
	return absPath, nil
}
