// Copyright © 2018 Banzai Cloud
//
// 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 cluster

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	"emperror.dev/errors"
	"github.com/Masterminds/semver/v3"
	"github.com/golang/protobuf/ptypes/duration"
	"github.com/jinzhu/copier"
	"github.com/jinzhu/gorm"
	"github.com/sirupsen/logrus"
	"golang.org/x/net/context"
	googleAuth "golang.org/x/oauth2/google"
	gkeCompute "google.golang.org/api/compute/v1"
	gke "google.golang.org/api/container/v1"
	"google.golang.org/api/googleapi"
	v1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"

	//  required by GCP authentication at runtime
	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd/api"

	"github.com/banzaicloud/pipeline/internal/cluster/clusteradapter/clustermodel"
	"github.com/banzaicloud/pipeline/internal/global"
	"github.com/banzaicloud/pipeline/internal/providers/google"
	"github.com/banzaicloud/pipeline/internal/secret/secrettype"
	pkgCluster "github.com/banzaicloud/pipeline/pkg/cluster"
	pkgClusterGoogle "github.com/banzaicloud/pipeline/pkg/cluster/gke"
	pkgCommon "github.com/banzaicloud/pipeline/pkg/common"
	pkgErrors "github.com/banzaicloud/pipeline/pkg/errors"
	pkgProviderGoogle "github.com/banzaicloud/pipeline/pkg/providers/google"
	"github.com/banzaicloud/pipeline/src/model"
	"github.com/banzaicloud/pipeline/src/secret"
	"github.com/banzaicloud/pipeline/src/utils"
)

const (
	statusRunning = "RUNNING"
	statusDone    = "DONE"
	statusError   = "ERROR"
)

const (
	adminSANamespace = "kube-system"
	adminSAName      = "pipeline"
	adminSARoleName  = "cluster-admin"
)

// constants to find Kubernetes resources
const (
	kubernetesIO   = "kubernetes.io"
	targetPrefix   = "gke-"
	clusterNameKey = "cluster-name"
)

// CreateGKEClusterFromRequest creates ClusterModel struct from the request
func CreateGKEClusterFromRequest(request *pkgCluster.CreateClusterRequest, orgID uint, userID uint) (*GKECluster, error) {
	c := GKECluster{
		log: log.WithFields(logrus.Fields{"cluster": request.Name, "organization": orgID}),
	}

	var err error
	c.repository, err = NewDBGKEClusterRepository(global.DB())
	if err != nil {
		return nil, errors.WrapIf(err, "failed to create GKE cluster repository")
	}

	nodePools, err := createNodePoolsModelFromRequest(request.Properties.CreateClusterGKE.NodePools, userID)
	if err != nil {
		return nil, err
	}

	c.model = &google.GKEClusterModel{
		Cluster: clustermodel.ClusterModel{
			Name:           request.Name,
			Location:       request.Location,
			OrganizationID: orgID,
			SecretID:       request.SecretId,
			Cloud:          google.Provider,
			Distribution:   google.ClusterDistributionGKE,
			CreatedBy:      userID,
		},

		MasterVersion: request.Properties.CreateClusterGKE.Master.Version,
		NodeVersion:   request.Properties.CreateClusterGKE.NodeVersion,
		NodePools:     nodePools,
		ProjectId:     request.Properties.CreateClusterGKE.ProjectId,
		Vpc:           request.Properties.CreateClusterGKE.Vpc,
		Subnet:        request.Properties.CreateClusterGKE.Subnet,
	}

	return &c, nil
}

type dbGKEClusterRepository struct {
	db *gorm.DB
}

func (r dbGKEClusterRepository) DeleteClusterModel(model *clustermodel.ClusterModel) error {
	return r.db.Delete(model).Error
}

func (r dbGKEClusterRepository) DeleteModel(model *google.GKEClusterModel) error {
	return r.db.Delete(model).Error
}

func (r dbGKEClusterRepository) DeleteNodePool(model *google.GKENodePoolModel) error {
	return r.db.Delete(model).Error
}

func (r dbGKEClusterRepository) SaveModel(model *google.GKEClusterModel) error {
	return r.db.Save(model).Error
}

func (r dbGKEClusterRepository) SaveStatusHistory(model *clustermodel.StatusHistoryModel) error {
	return r.db.Save(model).Error
}

// NewDBGKEClusterRepository returns a new GKEClusterRepository backed by a GORM DB
func NewDBGKEClusterRepository(db *gorm.DB) (GKEClusterRepository, error) {
	if db == nil {
		return nil, errors.New("db parameter cannot be nil")
	}
	return dbGKEClusterRepository{
		db: db,
	}, nil
}

// GKEClusterRepository describes a GKE cluster's persistent storage repository
type GKEClusterRepository interface {
	DeleteClusterModel(model *clustermodel.ClusterModel) error
	DeleteModel(model *google.GKEClusterModel) error
	DeleteNodePool(model *google.GKENodePoolModel) error
	SaveModel(model *google.GKEClusterModel) error
	SaveStatusHistory(model *clustermodel.StatusHistoryModel) error
}

// GKECluster struct for GKE cluster
type GKECluster struct {
	repository    GKEClusterRepository
	model         *google.GKEClusterModel
	googleCluster *gke.Cluster // Don't use this directly
	APIEndpoint   string
	log           logrus.FieldLogger
	CommonClusterBase
}

// GetOrganizationId gets org where the cluster belongs
func (c *GKECluster) GetOrganizationId() uint {
	return c.model.Cluster.OrganizationID
}

// GetLocation gets where the cluster is.
func (c *GKECluster) GetLocation() string {
	return c.model.Cluster.Location
}

// GetSecretId retrieves the secret id
func (c *GKECluster) GetSecretId() string {
	return c.model.Cluster.SecretID
}

// GetSshSecretId retrieves the secret id
func (c *GKECluster) GetSshSecretId() string {
	return c.model.Cluster.SSHSecretID
}

// SaveSshSecretId saves the ssh secret id to database
func (c *GKECluster) SaveSshSecretId(sshSecretId string) error {
	c.model.Cluster.SSHSecretID = sshSecretId

	err := c.repository.SaveModel(c.model)
	if err != nil {
		return errors.Wrap(err, "failed to save ssh secret id")
	}

	return nil
}

// GetGoogleCluster returns with a Cluster from GKE
func (c *GKECluster) GetGoogleCluster() (*gke.Cluster, error) {
	if c.googleCluster != nil {
		return c.googleCluster, nil
	}
	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return nil, err
	}

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return nil, err
	}

	cc := googleCluster{
		Name:      c.model.Cluster.Name,
		ProjectID: secretItem.Values[secrettype.ProjectId],
		Zone:      c.model.Cluster.Location,
	}
	cluster, err := getClusterGoogle(svc, cc)
	if err != nil {
		return nil, err
	}
	c.googleCluster = cluster
	return c.googleCluster, nil
}

// GetAPIEndpoint returns the Kubernetes Api endpoint
func (c *GKECluster) GetAPIEndpoint() (string, error) {
	if c.APIEndpoint != "" {
		return c.APIEndpoint, nil
	}
	cluster, err := c.GetGoogleCluster()
	if err != nil {
		return "", err
	}
	c.APIEndpoint = cluster.Endpoint
	return c.APIEndpoint, nil
}

// CreateCluster creates a new cluster
func (c *GKECluster) CreateCluster() error {
	c.log.Info("Start create cluster (Google)")

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return errors.WrapIf(err, "failed to retrieve cluster credential secret")
	}

	if c.model.ProjectId == "" {
		// if there's no projectId saved with the cluster take it from the secret
		c.model.ProjectId = secretItem.Values[secrettype.ProjectId]
	}

	// set region
	c.model.Region, err = c.getRegionByZone(c.model.ProjectId, c.model.Cluster.Location)
	if err != nil {
		c.log.Warnf("error during getting region: %s", err.Error())
	}

	c.model.Cluster.RbacEnabled = true

	log := c.log.WithFields(logrus.Fields{"project": c.model.ProjectId, "zone": c.model.Cluster.Location})

	log.Debug("Get Google Service Client")
	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return errors.WrapIf(err, "getting gke service client failed")
	}

	log.Info("Get Google Service Client succeeded")

	nodePools, err := createNodePoolsFromClusterModel(c.model)
	if err != nil {
		return errors.WrapIf(err, "create node pools from model failed")
	}

	cc := googleCluster{
		ProjectID:     c.model.ProjectId,
		Zone:          c.model.Cluster.Location,
		Name:          c.model.Cluster.Name,
		MasterVersion: c.model.MasterVersion,
		LegacyAbac:    false,
		Network:       c.model.Vpc,
		SubNetwork:    c.model.Subnet,
		NodePools:     nodePools,
	}

	ccr := generateClusterCreateRequest(cc)

	log.Debugf("cluster create request: %v", ccr)
	createCall, err := svc.Projects.Zones.Clusters.Create(cc.ProjectID, cc.Zone, ccr).Context(context.Background()).Do()
	if err != nil {
		if gErr, ok := err.(*googleapi.Error); ok && gErr.Code == http.StatusConflict {
			// cluster already exists. check if it was created by Pipeline
			gkeCluster, err := getClusterGoogle(svc, cc)
			if err != nil {
				return errors.WrapIf(err, "couldn't retrieve cluster details from cloud provider")
			}

			if !hasTag(gkeCluster, global.ManagedByPipelineTag, global.ManagedByPipelineValue) {
				return errors.Errorf("cluster %q already exists", gkeCluster.Name)
			}
		} else {
			return errors.WrapIf(err, "submitting cluster create request failed")
		}
	}

	if createCall != nil {
		log.Infof("cluster %s create is called for project %s and zone %s", cc.Name, cc.ProjectID, cc.Zone)
		log.Info("waiting for cluster...")

		if err := waitForOperation(newContainerOperation(svc, c.model.ProjectId, c.model.Cluster.Location), createCall.Name, log); err != nil {
			return errors.WrapIf(err, "waiting for cluster creation to complete failed")
		}
	} else {
		log.Info("cluster already created")
	}

	gkeCluster, err := getClusterGoogle(svc, cc)
	if err != nil {
		return errors.WrapIf(err, "couldn't retrieve cluster status")
	}
	if gkeCluster.Status == statusError {
		return errors.New(gkeCluster.StatusMessage)
	}

	c.googleCluster = gkeCluster

	c.updateCurrentVersions(gkeCluster)

	return nil
}

func (c *GKECluster) updateCurrentVersions(gkeCluster *gke.Cluster) {
	c.model.MasterVersion = gkeCluster.CurrentMasterVersion
	if len(gkeCluster.NodePools) != 0 && gkeCluster.NodePools[0] != nil {
		// currently we didn't support different node versions
		c.model.NodeVersion = gkeCluster.NodePools[0].Version
	}
}

// Persist save the cluster model
func (c *GKECluster) Persist() error {
	return errors.WrapIf(c.repository.SaveModel(c.model), "failed to persist cluster")
}

// DownloadK8sConfig downloads the kubeconfig file from cloud
func (c *GKECluster) DownloadK8sConfig() ([]byte, error) {
	config, err := c.getGoogleKubernetesConfig()
	if err != nil {
		return nil, errors.WrapIf(err, "retrieving kubernetes config failed")
	}
	// get config succeeded
	c.log.Info("Get k8s config succeeded")

	return config, nil
}

// GetName returns the name of the cluster
func (c *GKECluster) GetName() string {
	return c.model.Cluster.Name
}

// GetCloud returns the cloud type of the cluster
func (c *GKECluster) GetCloud() string {
	return c.model.Cluster.Cloud
}

// GetDistribution returns the distribution type of the cluster
func (c *GKECluster) GetDistribution() string {
	return c.model.Cluster.Distribution
}

// GetStatus gets cluster status
func (c *GKECluster) GetStatus() (*pkgCluster.GetClusterStatusResponse, error) {
	// c.log.Info("Create cluster status response")

	var hasSpotNodePool bool

	nodePools := make(map[string]*pkgCluster.NodePoolStatus)
	for _, np := range c.model.NodePools {
		if np != nil {
			nodePools[np.Name] = &pkgCluster.NodePoolStatus{
				Autoscaling:       np.Autoscaling,
				Preemptible:       np.Preemptible,
				Count:             np.NodeCount,
				InstanceType:      np.NodeInstanceType,
				MinCount:          np.NodeMinCount,
				MaxCount:          np.NodeMaxCount,
				Version:           c.model.NodeVersion,
				CreatorBaseFields: *NewCreatorBaseFields(np.CreatedAt, np.CreatedBy),
				Labels:            np.Labels,
			}
			if np.Preemptible {
				hasSpotNodePool = true
			}
		}
	}

	return &pkgCluster.GetClusterStatusResponse{
		Status:            c.model.Cluster.Status,
		StatusMessage:     c.model.Cluster.StatusMessage,
		Name:              c.model.Cluster.Name,
		Location:          c.model.Cluster.Location,
		Cloud:             c.model.Cluster.Cloud,
		Distribution:      c.model.Cluster.Distribution,
		Spot:              hasSpotNodePool,
		ResourceID:        c.model.Cluster.ID,
		Version:           c.model.MasterVersion,
		NodePools:         nodePools,
		CreatorBaseFields: *NewCreatorBaseFields(c.model.Cluster.CreatedAt, c.model.Cluster.CreatedBy),
		Region:            c.model.Region,
		StartedAt:         c.model.Cluster.StartedAt,
	}, nil
}

// DeleteCluster deletes cluster from google
func (c *GKECluster) DeleteCluster() error {
	c.log.Info("start delete gke cluster")

	if c == nil {
		return pkgErrors.ErrorNilCluster
	}

	if c.model.ProjectId == "" {
		// if there's no projectid saved with the cluster, take it from the secret
		secretItem, err := c.GetSecretWithValidation()
		if err != nil {
			return err
		}
		c.model.ProjectId = secretItem.Values[secrettype.ProjectId]
	}

	gkec := googleCluster{
		ProjectID: c.model.ProjectId,
		Name:      c.model.Cluster.Name,
		Zone:      c.model.Cluster.Location,
	}

	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return errors.WrapIf(err, "creating google service client failed")
	}

	// verify if cluster is managed by Pipeline
	gkeCluster, err := getClusterGoogle(svc, gkec)
	if err != nil {
		if gErr, ok := err.(*googleapi.Error); ok && gErr.Code == http.StatusNotFound {
			c.log.Info("cluster to be deleted doesn't exist")
			return nil
		}
		return errors.WrapIf(err, "couldn't retrieve cluster details")
	}

	if gkeCluster != nil && !hasTag(gkeCluster, global.ManagedByPipelineTag, global.ManagedByPipelineValue) {
		return errors.Errorf("cluster %q not managed by Pipeline", gkeCluster.Name)
	}

	if err := c.waitForResourcesDelete(); err != nil {
		return err
	}

	if err := c.callDeleteCluster(svc, &gkec); err != nil {
		return err
	}
	os.RemoveAll(gkec.TempCredentialPath)
	c.log.Info("delete succeeded")
	return nil
}

const (
	gkeResourceDeleteMaxAttempts = 12
	gkeResourceDeleteSleep       = 5 * time.Second
)

// waitForResourcesDelete waits until the Kubernetes destroys all the resources which it had created
func (c *GKECluster) waitForResourcesDelete() error {
	log := c.log.WithFields(logrus.Fields{"zone": c.model.Cluster.Location})

	log.Info("waiting for deleting cluster resources")

	csv, err := c.getComputeService()
	if err != nil {
		return errors.Wrap(err, "error during creating compute service")
	}

	project, err := c.GetProjectId()
	if err != nil {
		return errors.Wrap(err, "error during getting project id")
	}

	clusterName := c.model.Cluster.Name
	zone := c.model.Cluster.Location
	region, err := findRegionByZone(csv, project, zone)
	if err != nil {
		return errors.Wrap(err, "error during finding region by zone")
	}

	regionName := region.Name
	log.Infof("region name: %s", regionName)

	lb := newLoadBalancerHelper(csv, project, regionName, zone, clusterName)

	checkers := resourceCheckers{
		newFirewallChecker(csv, project, clusterName),
		newForwardingRulesChecker(csv, project, regionName, lb),
		newTargetPoolsChecker(csv, project, clusterName, regionName, zone, lb),
	}

	err = checkResources(checkers, gkeResourceDeleteMaxAttempts, gkeResourceDeleteSleep)
	if err != nil {
		return errors.Wrap(err, "error during checking resources")
	}

	return nil
}

// findRegionByZone returns region by zone
func findRegionByZone(csv *gkeCompute.Service, project, zone string) (*gkeCompute.Region, error) {
	regions, err := listRegions(csv, project)
	if err != nil {
		return nil, err
	}

	for _, r := range regions {
		if r != nil {
			for _, z := range r.Zones {
				if z == getZoneScope(project, zone) {
					return r, nil
				}
			}
		}
	}
	return nil, fmt.Errorf("cannot find zone[%s] in regions", zone)
}

func getZoneScope(project, zone string) string {
	return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s", project, zone)
}

// listRegions returns all region in project
func listRegions(csv *gkeCompute.Service, project string) ([]*gkeCompute.Region, error) {
	regionList, err := csv.Regions.List(project).Context(context.Background()).Do()
	if err != nil {
		return nil, err
	}
	return regionList.Items, nil
}

func (c *GKECluster) getRegionByZone(project string, zone string) (string, error) {
	c.log.Infof("start getting region by zone[%s]", zone)
	csv, err := c.getComputeService()
	if err != nil {
		return "", errors.Wrap(err, "error during creating compute service")
	}

	regions, err := csv.Regions.List(project).Context(context.Background()).Do()
	if err != nil {
		return "", errors.Wrap(err, "error during listing regions")
	}

	for _, i := range regions.Items {
		for _, z := range i.Zones {
			zoneScope := getZoneScope(project, zone)
			if z == zoneScope {
				log.Infof("match region: %s", i.Name)
				return i.Name, nil
			}
		}
	}

	return "", errors.Errorf("there's no zone [%s] in regions", zone)
}

// checkResources checks all load balancer resources deleted by Kubernetes
func checkResources(checkers resourceCheckers, maxAttempts int, sleep time.Duration) error {
	for _, rc := range checkers {
		log := log.WithFields(logrus.Fields{"type": rc.getType()})

		log.Debug("list resources")

		resources, err := rc.list()
		if err != nil {
			return err
		}

		log.Debugf("resource count: %d", len(resources))

		for _, resource := range resources {
			log := log.WithFields(logrus.Fields{"resource": resource, "type": rc.getType()})

			attempt := 0
			deleted := false

			for attempt <= maxAttempts && !deleted {
				log.Debugf("waiting for resource to be deleted %d/%d", attempt, maxAttempts)
				err := rc.isResourceDeleted(resource)
				if err == nil {
					log.Info("resource deleted")
					deleted = true
					break
				} else {
					log.Warn(err.Error())
					time.Sleep(sleep)
				}
				attempt++
			}

			if !deleted {
				log.Info("force delete")
				if err := rc.forceDelete(resource); err != nil {
					return err
				}
			}
		}
	}

	return nil
}

// UpdateNodePools updates nodes pools of a cluster
func (c *GKECluster) UpdateNodePools(request *pkgCluster.UpdateNodePoolsRequest, userId uint) error {
	c.log.Info("Start updating cluster (gke) nodepools")

	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return errors.WrapIf(err, "Unable to retrieve GoogleServiceClient")
	}

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return errors.WrapIf(err, "Unable to retrieve secret")
	}

	projectId := secretItem.Values[secrettype.ProjectId]

	// Update node pools
	for poolName, nodePool := range request.NodePools {
		log.Infof("Updating node size to %v for node pool %s", nodePool.Count, poolName)
		updateCall, err := svc.Projects.Zones.Clusters.NodePools.SetSize(projectId, c.model.Cluster.Location, c.GetName(), poolName, &gke.SetNodePoolSizeRequest{
			NodeCount: int64(nodePool.Count),
		}).Context(context.Background()).Do()
		if err != nil {
			return errors.WrapIf(err, "Updating nodepool size failed")
		}

		if err = waitForOperation(newContainerOperation(svc, projectId, c.model.Cluster.Location), updateCall.Name, log); err != nil {
			return errors.WrapIf(err, "Error while waiting for nodepool size update")
		}

		log.Infof("Node pool %s size change is called for project %s, zone %s and cluster %s. Status Code %v", poolName, projectId, c.model.Cluster.Location, c.GetName(), updateCall.HTTPStatusCode)
		c.setNodePoolSize(poolName, nodePool.Count)
	}
	return nil
}

// UpdateCluster updates GKE cluster in cloud
func (c *GKECluster) UpdateCluster(updateRequest *pkgCluster.UpdateClusterRequest, userId uint) error {
	c.log.Info("Start updating cluster (gke)")

	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return err
	}

	updateNodePoolsModel, err := createNodePoolsModelFromRequest(updateRequest.GKE.NodePools, userId)
	if err != nil {
		return err
	}

	googleClusterModel := &google.GKEClusterModel{}

	copier.Copy(googleClusterModel, c.model) // nolint: errcheck
	googleClusterModel.NodePools = updateNodePoolsModel

	googleClusterModel.NodeVersion = updateRequest.GKE.NodeVersion
	googleClusterModel.MasterVersion = updateRequest.GKE.Master.Version

	updatedNodePools, err := createNodePoolsFromClusterModel(googleClusterModel)
	if err != nil {
		return err
	}

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return err
	}

	projectId := secretItem.Values[secrettype.ProjectId]

	cc := googleCluster{
		Name:          c.model.Cluster.Name,
		ProjectID:     projectId,
		Zone:          c.model.Cluster.Location,
		MasterVersion: updateRequest.GKE.Master.Version,
		NodePools:     updatedNodePools,
	}

	res, err := callUpdateClusterGoogle(svc, cc, c.model.Cluster.Location, projectId)
	if err != nil {
		be := getBanzaiErrorFromError(err)
		// TODO status code !?
		return errors.New(be.Message)
	}
	c.log.Info("Cluster update succeeded")
	c.googleCluster = res

	c.updateCurrentVersions(res)

	// update model to save
	c.updateModel(res, updatedNodePools)

	return nil
}

func (c *GKECluster) updateModel(cluster *gke.Cluster, updatedNodePools []*gke.NodePool) {
	// Update the model from the cluster data read back from Google
	c.model.MasterVersion = cluster.CurrentMasterVersion
	c.model.NodeVersion = cluster.CurrentNodeVersion

	var newNodePoolsModels []*google.GKENodePoolModel
	for _, clusterNodePool := range cluster.NodePools {
		updated := false

		for _, nodePoolModel := range c.model.NodePools {
			if clusterNodePool.Name == nodePoolModel.Name {
				nodePoolModel.NodeInstanceType = clusterNodePool.Config.MachineType

				if clusterNodePool.Autoscaling != nil {
					nodePoolModel.Autoscaling = clusterNodePool.Autoscaling.Enabled
					nodePoolModel.NodeMinCount = int(clusterNodePool.Autoscaling.MinNodeCount)
					nodePoolModel.NodeMaxCount = int(clusterNodePool.Autoscaling.MaxNodeCount)
				}

				// TODO: This is ugly but Google API doesn't expose the current node count for a node pool
				for _, updatedNodePool := range updatedNodePools {
					if updatedNodePool.Name == clusterNodePool.Name {
						nodePoolModel.NodeCount = int(updatedNodePool.InitialNodeCount)

						break
					}
				}

				updated = true
				break
			}
		}

		if !updated {
			nodePoolModelAdd := &google.GKENodePoolModel{
				Name:             clusterNodePool.Name,
				NodeInstanceType: clusterNodePool.Config.MachineType,
				NodeCount:        int(clusterNodePool.InitialNodeCount),
			}
			if clusterNodePool.Autoscaling != nil {
				nodePoolModelAdd.Autoscaling = clusterNodePool.Autoscaling.Enabled
				nodePoolModelAdd.NodeMinCount = int(clusterNodePool.Autoscaling.MinNodeCount)
				nodePoolModelAdd.NodeMaxCount = int(clusterNodePool.Autoscaling.MaxNodeCount)
			}

			newNodePoolsModels = append(newNodePoolsModels, nodePoolModelAdd)
		}
	}

	for _, newNodePoolModel := range newNodePoolsModels {
		c.model.NodePools = append(c.model.NodePools, newNodePoolModel)
	}

	// mark for deletion the node pool model entries that has no corresponding node pool in the cluster
	for _, nodePoolModel := range c.model.NodePools {
		found := false

		for _, clusterNodePool := range cluster.NodePools {
			if nodePoolModel.Name == clusterNodePool.Name {
				found = true
				break
			}
		}

		if !found {
			nodePoolModel.Delete = true
		}
	}
}

// GetID returns the specified cluster id
func (c *GKECluster) GetID() uint {
	return c.model.Cluster.ID
}

func (c *GKECluster) GetUID() string {
	return c.model.Cluster.UID
}

func (c *GKECluster) getGoogleServiceClient() (*gke.Service, error) {
	client, err := c.newClientFromCredentials()
	if err != nil {
		return nil, errors.WrapIf(err, "retrieving cluster credentials secret failed")
	}

	// New client from credentials
	return gke.New(client)
}

// GKE cluster to google calls
type googleCluster struct {
	// ProjectID is the ID of your project to use when creating a cluster
	ProjectID string `json:"projectId,omitempty"`
	// The zone to launch the cluster
	Zone string
	// The IP address range of the container pods
	ClusterIpv4Cidr string
	// An optional description of this cluster
	Description string

	// the kubernetes master version
	MasterVersion string
	// The authentication information for accessing the master
	MasterAuth *gke.MasterAuth

	// The name of this cluster
	Name string
	// The path to the credential file(key.json)
	CredentialPath string
	// The content of the credential
	CredentialContent string
	// the temp file of the credential
	TempCredentialPath string
	// Enable alpha feature
	EnableAlphaFeature bool
	// Configuration for the HTTP (L7) load balancing controller addon
	HTTPLoadBalancing bool
	// Configuration for the horizontal pod autoscaling feature, which increases or decreases the number of replica pods a replication controller has based on the resource usage of the existing pods
	HorizontalPodAutoscaling bool
	// Configuration for the Kubernetes Dashboard
	KubernetesDashboard bool
	// Configuration for NetworkPolicy
	NetworkPolicyConfig bool
	// The list of Google Compute Engine locations in which the cluster's nodes should be located
	Locations []string
	// Network
	Network string
	// Sub Network
	SubNetwork string
	// Configuration for LegacyAbac
	LegacyAbac bool
	// Image Type
	ImageType string
	// The node pools the cluster's nodes are created from
	NodePools []*gke.NodePool
}

func generateClusterCreateRequest(cc googleCluster) *gke.CreateClusterRequest {
	request := gke.CreateClusterRequest{
		Cluster: &gke.Cluster{},
	}
	request.Cluster.Name = cc.Name
	request.Cluster.Zone = cc.Zone
	request.Cluster.InitialClusterVersion = cc.MasterVersion
	request.Cluster.ClusterIpv4Cidr = cc.ClusterIpv4Cidr
	request.Cluster.Description = cc.Description
	request.Cluster.EnableKubernetesAlpha = cc.EnableAlphaFeature
	request.Cluster.LoggingService = "none"
	request.Cluster.MonitoringService = "none"
	request.Cluster.AddonsConfig = &gke.AddonsConfig{
		HttpLoadBalancing:        &gke.HttpLoadBalancing{Disabled: !cc.HTTPLoadBalancing},
		HorizontalPodAutoscaling: &gke.HorizontalPodAutoscaling{Disabled: !cc.HorizontalPodAutoscaling},
		KubernetesDashboard:      &gke.KubernetesDashboard{Disabled: !cc.KubernetesDashboard},
		// 	NetworkPolicyConfig:      &gke.NetworkPolicyConfig{Disabled: !cc.NetworkPolicyConfig},
	}
	request.Cluster.Network = cc.Network
	request.Cluster.Subnetwork = cc.SubNetwork
	request.Cluster.LegacyAbac = &gke.LegacyAbac{
		Enabled: cc.LegacyAbac,
	}
	request.Cluster.MasterAuth = &gke.MasterAuth{}
	request.Cluster.NodePools = cc.NodePools
	request.Cluster.ResourceLabels = pipelineTags()
	request.ProjectId = cc.ProjectID

	return &request
}

// pipelineTags returns resource tags based on the pipeline uuid if available
func pipelineTags() map[string]string {
	value := global.Config.Pipeline.UUID

	tags := map[string]string{
		global.ManagedByPipelineTag: global.ManagedByPipelineValue,
	}
	if value != "" {
		tags[global.ManagedByPipelineUUIDTag] = value
	}

	return tags
}

func getBanzaiErrorFromError(err error) *pkgCommon.BanzaiResponse {
	if err == nil {
		// error is nil
		return &pkgCommon.BanzaiResponse{
			StatusCode: http.StatusInternalServerError,
		}
	}

	googleErr, ok := errors.Cause(err).(*googleapi.Error)
	if ok {
		// error is googleapi error
		return &pkgCommon.BanzaiResponse{
			StatusCode: googleErr.Code,
			Message:    googleErr.Message,
		}
	}

	// default
	return &pkgCommon.BanzaiResponse{
		StatusCode: http.StatusInternalServerError,
		Message:    err.Error(),
	}
}

func getClusterGoogle(svc *gke.Service, cc googleCluster) (*gke.Cluster, error) {
	return svc.Projects.Zones.Clusters.Get(cc.ProjectID, cc.Zone, cc.Name).Context(context.TODO()).Do()
}

func (c *GKECluster) callDeleteCluster(svc *gke.Service, cc *googleCluster) error {
	c.log.Infof("removing cluster %v from project %v, zone %v", cc.Name, cc.ProjectID, cc.Zone)
	deleteCall, err := svc.Projects.Zones.Clusters.Delete(cc.ProjectID, cc.Zone, cc.Name).Context(context.Background()).Do()

	if err != nil {
		if gErr, ok := err.(*googleapi.Error); ok && gErr.Code == http.StatusNotFound {
			// cluster has already been deleted from cloud provider
			c.log.Info("cluster to be deleted doesn't exist")
		} else {
			return errors.WrapIf(err, "initiating cluster deletion failed")
		}
	} else {
		c.log.Debugf("cluster delete is called. Status Code %v", deleteCall.HTTPStatusCode)

		if deleteCall != nil {
			c.log.Info("waiting for cluster deletion to complete...")

			if err := waitForOperation(newContainerOperation(svc, c.model.ProjectId, c.model.Cluster.Location), deleteCall.Name, c.log); err != nil {
				return errors.WrapIf(err, "waiting for cluster deletion to complete failed")
			}
		}
	}

	return nil
}

func callUpdateClusterGoogle(svc *gke.Service, cc googleCluster, location, projectId string) (*gke.Cluster, error) {
	log := log.WithField("cluster", cc.Name)

	log.Infof("Updating cluster: %#v", cc)

	updatedCluster, err := getClusterGoogle(svc, cc)
	if err != nil {
		return nil, err
	}

	// master version reported by GCP
	currentMasterVersion, err := semver.NewVersion(updatedCluster.CurrentMasterVersion)
	if err != nil {
		return nil, errors.WrapIf(err, "couldn't parse current master version")
	}

	var desiredMasterVersion *semver.Version
	if cc.MasterVersion != "" {
		desiredMasterVersion, err = semver.NewVersion(cc.MasterVersion)
		if err != nil {
			return nil, errors.WrapIf(err, "couldn't parse desired master version")
		}
	}

	if desiredMasterVersion != nil && desiredMasterVersion.GreaterThan(currentMasterVersion) {
		log.Infof("Updating master to %v version", cc.MasterVersion)
		updateCall, err := svc.Projects.Zones.Clusters.Update(cc.ProjectID, cc.Zone, cc.Name, &gke.UpdateClusterRequest{
			Update: &gke.ClusterUpdate{
				DesiredMasterVersion: cc.MasterVersion,
			},
		}).Context(context.Background()).Do()
		if err != nil {
			return nil, err
		}
		log.Infof("Cluster %s update is called for project %s and zone %s. Status Code %v", cc.Name, cc.ProjectID, cc.Zone, updateCall.HTTPStatusCode)
		if err = waitForOperation(newContainerOperation(svc, projectId, location), updateCall.Name, log); err != nil {
			return nil, err
		}

		updatedCluster, err = getClusterGoogle(svc, cc)
		if err != nil {
			return nil, err
		}
	}

	// Collect node pools that have to be deleted and delete them before
	// resizing exiting ones or creating new ones to minimize tha chance
	// of hitting quota limits
	var nodePoolsToDelete []string
	for _, currentClusterNodePool := range updatedCluster.NodePools {
		var i int
		for i = 0; i < len(cc.NodePools); i++ {
			if currentClusterNodePool.Name == cc.NodePools[i].Name {
				break
			}
		}

		if i == len(cc.NodePools) {
			// cluster node pool with given name not found in the update request thus we need to delete it
			nodePoolsToDelete = append(nodePoolsToDelete, currentClusterNodePool.Name)
		}
	}

	var nodePoolsToCreate []*gke.NodePool
	for _, nodePoolFromUpdReq := range cc.NodePools {
		var i int
		for i = 0; i < len(updatedCluster.NodePools); i++ {
			if nodePoolFromUpdReq.Name == updatedCluster.NodePools[i].Name {
				break
			}
		}
		if i == len(updatedCluster.NodePools) {
			nodePoolsToCreate = append(nodePoolsToCreate, nodePoolFromUpdReq)
		}
	}

	// Update node pools
	for _, nodePool := range cc.NodePools {
		for i := 0; i < len(updatedCluster.NodePools); i++ {
			if updatedCluster.NodePools[i].Name == nodePool.Name {
				var desiredNodePoolVersion *semver.Version
				if nodePool.Version != "" {
					desiredNodePoolVersion, err = semver.NewVersion(nodePool.Version)
					if err != nil {
						return nil, errors.WrapIfWithDetails(err, "couldn't parse desired node pool version", "node pool", nodePool.Name)
					}
				}
				currentNodePoolVersion, err := semver.NewVersion(updatedCluster.NodePools[i].Version)
				if err != nil {
					return nil, errors.WrapIfWithDetails(err, "couldn't parse current node pool version", "node pool", nodePool.Name)
				}

				if desiredNodePoolVersion != nil && desiredNodePoolVersion.GreaterThan(currentNodePoolVersion) {
					log.Infof("Updating node pool %s to %v version", nodePool.Name, nodePool.Version)
					updateCall, err := svc.Projects.Zones.Clusters.NodePools.Update(cc.ProjectID, cc.Zone, cc.Name, nodePool.Name, &gke.UpdateNodePoolRequest{
						NodeVersion: nodePool.Version,
					}).Context(context.Background()).Do()
					if err != nil {
						return nil, err
					}
					log.Infof("Node pool %s update is called for project %s, zone %s and cluster %s. Status Code %v", nodePool.Name, cc.ProjectID, cc.Zone, cc.Name, updateCall.HTTPStatusCode)
					if err := waitForOperation(newContainerOperation(svc, projectId, location), updateCall.Name, log); err != nil {
						return nil, err
					}
				}

				if autoscalingHasBeenUpdated(nodePool, updatedCluster.NodePools[i]) {
					var err error

					autoScalingInput := &gke.SetNodePoolAutoscalingRequest{
						Autoscaling: &gke.NodePoolAutoscaling{
							Enabled: false,
						},
					}

					if nodePool.Autoscaling.Enabled {
						log.Infof("Updating node pool %s enable Autoscaling", nodePool.Name)
						autoScalingInput = &gke.SetNodePoolAutoscalingRequest{
							Autoscaling: &gke.NodePoolAutoscaling{
								Enabled:      true,
								MinNodeCount: nodePool.Autoscaling.MinNodeCount,
								MaxNodeCount: nodePool.Autoscaling.MaxNodeCount,
							},
						}
					} else {
						log.Infof("Updating node pool %s disable Autoscaling", nodePool.Name)
					}

					operation, err := svc.Projects.Zones.Clusters.NodePools.Autoscaling(cc.ProjectID, cc.Zone, cc.Name, nodePool.Name, autoScalingInput).Context(context.Background()).Do()
					if err != nil {
						return nil, err
					}

					log.Infof("Node pool %s update is called for project %s, zone %s and cluster %s", nodePool.Name, cc.ProjectID, cc.Zone, cc.Name)
					if err = waitForOperation(newContainerOperation(svc, projectId, location), operation.Name, log); err != nil {
						return nil, err
					}

					updatedCluster, err = getClusterGoogle(svc, cc)
					if err != nil {
						return nil, err
					}
				}

				if nodePool.InitialNodeCount > 0 && nodePool.InitialNodeCount != updatedCluster.NodePools[i].InitialNodeCount {
					log.Infof("Updating node size to %v for node pool %s", nodePool.InitialNodeCount, nodePool.Name)
					updateCall, err := svc.Projects.Zones.Clusters.NodePools.SetSize(cc.ProjectID, cc.Zone, cc.Name, nodePool.Name, &gke.SetNodePoolSizeRequest{
						NodeCount: nodePool.InitialNodeCount,
					}).Context(context.Background()).Do()
					if err != nil {
						return nil, err
					}
					log.Infof("Node pool %s size change is called for project %s, zone %s and cluster %s. Status Code %v", nodePool.Name, cc.ProjectID, cc.Zone, cc.Name, updateCall.HTTPStatusCode)
					if err = waitForOperation(newContainerOperation(svc, projectId, location), updateCall.Name, log); err != nil {
						return nil, err
					}

					updatedCluster, err = getClusterGoogle(svc, cc)
					if err != nil {
						return nil, err
					}
				}

				break
			}
		}
	}

	// Create node pools
	for _, nodePoolToCreate := range nodePoolsToCreate {
		log.Infof("Creating node pool %s", nodePoolToCreate.Name)

		createCall, err := svc.Projects.Zones.Clusters.NodePools.Create(cc.ProjectID, cc.Zone, cc.Name, &gke.CreateNodePoolRequest{
			NodePool: nodePoolToCreate,
		}).Context(context.Background()).Do()
		if err != nil {
			return nil, err
		}
		log.Infof("Node pool %s create is called for project %s, zone %s and cluster %s. Status Code %v", nodePoolToCreate.Name, cc.ProjectID, cc.Zone, cc.Name, createCall.HTTPStatusCode)
		if err = waitForOperation(newContainerOperation(svc, projectId, location), createCall.Name, log); err != nil {
			return nil, err
		}

		updatedCluster, err = getClusterGoogle(svc, cc)
		if err != nil {
			return nil, err
		}
	}

	// Delete node pools
	for _, nodePoolName := range nodePoolsToDelete {
		log.Infof("Deleting node pool %s", nodePoolName)

		deleteCall, err := svc.Projects.Zones.Clusters.NodePools.Delete(cc.ProjectID, cc.Zone, cc.Name, nodePoolName).Context(
			context.Background()).Do()
		if err != nil {
			return nil, err
		}
		log.Infof("Node pool %s delete is called for project %s, zone %s and cluster %s. Status Code %v", nodePoolName, cc.ProjectID, cc.Zone, cc.Name, deleteCall.HTTPStatusCode)
		if err = waitForOperation(newContainerOperation(svc, projectId, location), deleteCall.Name, log); err != nil {
			return nil, err
		}
		updatedCluster, err = getClusterGoogle(svc, cc)
		if err != nil {
			return nil, err
		}
	}

	return updatedCluster, nil
}

func autoscalingHasBeenUpdated(updatedNodePool *gke.NodePool, actualNodePool *gke.NodePool) bool {
	if actualNodePool.Autoscaling == nil {
		return updatedNodePool.Autoscaling.Enabled
	}
	if updatedNodePool.Autoscaling.Enabled && actualNodePool.Autoscaling.Enabled {
		if updatedNodePool.Autoscaling.MinNodeCount != actualNodePool.Autoscaling.MinNodeCount {
			return true
		}
		if updatedNodePool.Autoscaling.MaxNodeCount != actualNodePool.Autoscaling.MaxNodeCount {
			return true
		}
		return false
	} else if !updatedNodePool.Autoscaling.Enabled && !actualNodePool.Autoscaling.Enabled {
		return false
	}
	return true
}

func (c *GKECluster) getGoogleKubernetesConfig() ([]byte, error) {
	c.log.Info("Get Google Service Client")
	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return nil, errors.WrapIf(err, "creating google service client failed")
	}
	c.log.Info("Get Google Service Client succeeded")

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return nil, errors.WrapIf(err, "retrieving cluster credentials secret failed")
	}

	c.log.Infof("Get gke cluster with name %s", c.model.Cluster.Name)
	cl, err := getClusterGoogle(svc, googleCluster{
		Name:      c.model.Cluster.Name,
		ProjectID: secretItem.Values[secrettype.ProjectId],
		Zone:      c.model.Cluster.Location,
	})
	if err != nil {
		return nil, errors.WrapIf(err, "retrieving GKE cluster provider failed")
	}

	c.log.Info("Generate Service Account token")
	serviceAccountToken, err := c.generateServiceAccountTokenForGke(cl)
	if err != nil {
		return nil, errors.WrapIf(err, "getting service account for GKE cluster failed")
	}

	finalCl := pkgCluster.KubernetesCluster{
		ClientCertificate:   cl.MasterAuth.ClientCertificate,
		ClientKey:           cl.MasterAuth.ClientKey,
		RootCACert:          cl.MasterAuth.ClusterCaCertificate,
		Name:                cl.Name,
		Username:            cl.MasterAuth.Username,
		Password:            cl.MasterAuth.Password,
		Version:             cl.CurrentMasterVersion,
		Endpoint:            cl.Endpoint,
		NodeCount:           cl.CurrentNodeCount,
		Metadata:            map[string]string{},
		ServiceAccountToken: serviceAccountToken,
		Status:              cl.Status,
	}

	finalCl.Metadata["nodePools"] = fmt.Sprintf("%v", cl.NodePools)

	config, err := pkgCluster.StoreConfig(&finalCl)
	if err != nil {
		return nil, errors.WrapIf(err, "storing kubernetes config failed")
	}
	return config, nil
}

func (c *GKECluster) generateServiceAccountTokenForGke(cluster *gke.Cluster) (string, error) {
	capem, err := base64.StdEncoding.DecodeString(cluster.MasterAuth.ClusterCaCertificate)
	if err != nil {
		return "", errors.WrapIf(err, "base64 decode of ca certificate failed")
	}

	host := cluster.Endpoint
	if !strings.HasPrefix(host, "https://") {
		host = fmt.Sprintf("https://%s", host)
	}

	// generate GCP access token
	ctx := context.Background()
	clusterSecret, err := c.GetSecretWithValidation()
	if err != nil {
		return "", errors.WrapIf(err, "retrieving cluster credentials secret failed")
	}

	serviceAccount := pkgProviderGoogle.CreateServiceAccount(clusterSecret.Values)
	jsonConfig, err := json.Marshal(serviceAccount)
	if err != nil {
		return "", errors.WrapIf(err, "marshaling cluster credentials secret to json format failed")
	}

	googleCredentials, err := googleAuth.CredentialsFromJSON(ctx, jsonConfig, gke.CloudPlatformScope)
	if err != nil {
		return "", errors.WrapIf(err, "creating Google credentials failed")
	}

	iamSvc := pkgProviderGoogle.NewIamSvc(googleCredentials)

	// get short lived service account access credentials
	tokenResp, err := iamSvc.GenerateNewAccessToken(
		serviceAccount.ClientEmail,
		&duration.Duration{
			Seconds: 600, // token expires after 10 mins
		})
	if err != nil {
		return "", errors.WrapIf(err, "getting access token failed")
	}

	tokenExpiry := time.Unix(tokenResp.GetExpireTime().GetSeconds(), int64(tokenResp.GetExpireTime().GetNanos()))

	// kubernetes config using GCP authenticator
	config := &rest.Config{
		Host: host,
		TLSClientConfig: rest.TLSClientConfig{
			CAData: capem,
		},
		AuthProvider: &api.AuthProviderConfig{
			Name: "gcp",
			Config: map[string]string{
				"access-token": tokenResp.GetAccessToken(),
				"expiry":       tokenExpiry.Format(time.RFC3339Nano),
				"cmd-path":     "/bin/true",
			},
		},
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return "", errors.WrapIf(err, "creating kubernetes clientset failed")
	}

	return generateServiceAccountToken(clientset)
}

func generateServiceAccountToken(clientset *kubernetes.Clientset) (string, error) {
	serviceAccount := &v1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name: adminSAName,
		},
	}

	ctx := context.Background()

	_, err := clientset.CoreV1().ServiceAccounts(adminSANamespace).Create(ctx, serviceAccount, metav1.CreateOptions{})
	if err != nil && !k8sErrors.IsAlreadyExists(err) {
		return "", errors.WrapIfWithDetails(err, "creating service account failed", "namespace", adminSANamespace, "service account", serviceAccount)
	}

	adminRole := &rbacv1.ClusterRole{
		ObjectMeta: metav1.ObjectMeta{
			Name: adminSARoleName,
		},
		Rules: []rbacv1.PolicyRule{
			{
				APIGroups: []string{"*"},
				Resources: []string{"*"},
				Verbs:     []string{"*"},
			},
			{
				NonResourceURLs: []string{"*"},
				Verbs:           []string{"*"},
			},
		},
	}
	clusterAdminRole, err := clientset.RbacV1().ClusterRoles().Get(ctx, adminSARoleName, metav1.GetOptions{})
	if err != nil {
		clusterAdminRole, err = clientset.RbacV1().ClusterRoles().Create(ctx, adminRole, metav1.CreateOptions{})
		if err != nil {
			return "", errors.WrapIfWithDetails(err, "creating cluster role failed", "cluster role", adminRole.Name)
		}
	}

	clusterRoleBinding := &rbacv1.ClusterRoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name: fmt.Sprintf("%s-clusterRoleBinding", adminSAName),
		},
		Subjects: []rbacv1.Subject{
			{
				Kind:      "ServiceAccount",
				Name:      serviceAccount.Name,
				Namespace: adminSANamespace,
			},
		},
		RoleRef: rbacv1.RoleRef{
			Kind: "ClusterRole",
			Name: clusterAdminRole.Name,
		},
	}
	if _, err = clientset.RbacV1().ClusterRoleBindings().Create(ctx, clusterRoleBinding, metav1.CreateOptions{}); err != nil && !k8sErrors.IsAlreadyExists(err) {
		return "", errors.WrapIfWithDetails(err, "creating cluster role binding failed", "cluster role", clusterAdminRole.Name, "service account", serviceAccount.Name)
	}

	if serviceAccount, err = clientset.CoreV1().ServiceAccounts(adminSANamespace).Get(ctx, serviceAccount.Name, metav1.GetOptions{}); err != nil {
		return "", errors.WrapIfWithDetails(err, "retrieving service account failed", "namespace", adminSANamespace, "service account", serviceAccount.Name)
	}

	// Note: From K8s 1.24 ServiceAccounts don't have default secrets
	// created for them (source:
	// https://github.com/kubernetes/kubernetes/blob/master/CHANGELOG/CHANGELOG-1.24.md#no-really-you-must-read-this-before-you-upgrade,
	// second point). We create one manually to provide an SA token.
	var secretObj *v1.Secret

	secretName := adminSAName + "-token"
	if len(serviceAccount.Secrets) != 0 {
		secretName = serviceAccount.Secrets[0].Name
	}

	secretObj, err = clientset.CoreV1().Secrets(adminSANamespace).Get(ctx, secretName, metav1.GetOptions{})
	if err != nil &&
		k8sErrors.IsNotFound(err) {
		adminSATokenSecret := v1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      secretName,
				Namespace: adminSANamespace,
				Annotations: map[string]string{
					"kubernetes.io/service-account.name": adminSAName,
				},
			},
			Type: "kubernetes.io/service-account-token",
		}

		secretObj, err = clientset.CoreV1().Secrets(adminSANamespace).Create(ctx, &adminSATokenSecret, metav1.CreateOptions{})
		if err != nil {
			return "", errors.WrapIfWithDetails(err, "creating kubernetes secret failed", "namespace", adminSANamespace, "secret", secretName)
		}
	} else if err != nil { // && !k8sErrors.IsNotFound(err) {
		return "", errors.WrapIfWithDetails(
			err,
			"retrieving kubernetes secret failed with unexpected error",
			"namespace", adminSANamespace,
			"secret", secretName,
		)
	}

	token, ok := secretObj.Data["token"]
	if !ok {
		return "", errors.NewWithDetails("retrieving secret token data failed", "namespace", adminSANamespace, "secret", secretName, "key", "token")
	}

	return string(token), nil
}

// CreateGKEClusterFromModel creates ClusterModel struct from model
func CreateGKEClusterFromModel(clusterModel *model.ClusterModel) (*GKECluster, error) {
	db := global.DB()

	m := google.GKEClusterModel{
		ClusterID: clusterModel.ID,
	}

	err := db.Where(m).Preload("Cluster").Preload("NodePools").First(&m).Error
	if err != nil {
		return nil, err
	}

	repository, err := NewDBGKEClusterRepository(db)
	if err != nil {
		return nil, errors.WrapIf(err, "failed to create DB GKE cluster repository")
	}

	log := log.WithFields(logrus.Fields{"cluster": clusterModel.Name, "clusterID": m.Cluster.ID})

	gkeCluster := GKECluster{
		repository: repository,
		model:      &m,
		log:        log,
	}
	return &gkeCluster, nil
}

// AddDefaultsToUpdate adds defaults to update request
func (c *GKECluster) AddDefaultsToUpdate(r *pkgCluster.UpdateClusterRequest) {
	// TODO: error handling
	defGooglePools, _ := createNodePoolsFromClusterModel(c.model)

	defGoogleMaster := &pkgClusterGoogle.Master{
		Version: c.model.MasterVersion,
	}

	// ---- [ Node check ] ---- //
	if r.GKE.NodePools == nil {
		log.Warn("'nodePools' field is empty. Load it from stored data.")

		r.GKE.NodePools = make(map[string]*pkgClusterGoogle.NodePool)
		for _, nodePool := range defGooglePools {
			r.GKE.NodePools[nodePool.Name] = &pkgClusterGoogle.NodePool{
				Count:            int(nodePool.InitialNodeCount),
				NodeInstanceType: nodePool.Config.MachineType,
			}
			if nodePool.Autoscaling != nil {
				r.GKE.NodePools[nodePool.Name].Autoscaling = nodePool.Autoscaling.Enabled
				r.GKE.NodePools[nodePool.Name].MinCount = int(nodePool.Autoscaling.MinNodeCount)
				r.GKE.NodePools[nodePool.Name].MaxCount = int(nodePool.Autoscaling.MaxNodeCount)
			}
		}
	}
	// ---- [ Master check ] ---- //
	if r.GKE.Master == nil {
		log.Warn("'master' field is empty. Load it from stored data.")
		r.GKE.Master = defGoogleMaster
	}

	// ---- [ NodeCount check] ---- //
	for name, nodePoolData := range r.GKE.NodePools {
		if nodePoolData.Count == 0 {
			// initialize with count read from db
			var i int
			for i = 0; i < len(c.model.NodePools); i++ {
				if c.model.NodePools[i].Name == name {
					nodePoolData.Count = c.model.NodePools[i].NodeCount
					log.Warnf("Node count for node pool %s initiated from database to value: %d", name, nodePoolData.Count)
					break
				}
			}
			if i == len(c.model.NodePools) {
				// node pool not found in db; set count to default value
				nodePoolData.Count = pkgCommon.DefaultNodeMinCount
				log.Warnf("Node count for node pool %s set to default value: %d", name, nodePoolData.Count)
			}
		}
	}

	// ---- [ Node Version check] ---- //
	if len(r.GKE.NodeVersion) == 0 {
		nodeVersion := c.model.NodeVersion
		log.Warnf("Node K8s version: %s", nodeVersion)
		r.GKE.NodeVersion = nodeVersion
	}

	// ---- [ Master Version check] ---- //
	if len(r.GKE.Master.Version) == 0 {
		masterVersion := c.model.MasterVersion
		log.Warnf("Master K8s version: %s", masterVersion)
		r.GKE.Master.Version = masterVersion
	}
}

// CheckEqualityToUpdate validates the update request
func (c *GKECluster) CheckEqualityToUpdate(r *pkgCluster.UpdateClusterRequest) error {
	// create update request struct with the stored data to check equality
	nodePools, _ := createNodePoolsRequestDataFromNodePoolModel(c.model.NodePools)
	preCl := &pkgClusterGoogle.UpdateClusterGoogle{
		Master: &pkgClusterGoogle.Master{
			Version: c.model.MasterVersion,
		},
		NodeVersion: c.model.NodeVersion,
		NodePools:   nodePools,
	}

	c.log.Info("Check stored & updated cluster equals")

	// check equality
	return isDifferent(r.GKE, preCl)
}

// DeleteFromDatabase deletes model from the database
func (c *GKECluster) DeleteFromDatabase() error {
	if err := c.repository.DeleteClusterModel(&c.model.Cluster); err != nil {
		return err
	}

	for _, nodePool := range c.model.NodePools {
		if err := c.repository.DeleteNodePool(nodePool); err != nil {
			return err
		}
	}

	if err := c.repository.DeleteModel(c.model); err != nil {
		return err
	}

	c.model = nil

	return nil
}

// GetGkeServerConfig returns configuration info about the Kubernetes Engine service.
func (c *GKECluster) GetGkeServerConfig(zone string) (*gke.ServerConfig, error) {
	c.log.Info("Start getting configuration info")

	c.log.Info("Get Google service client")
	svc, err := c.getGoogleServiceClient()
	if err != nil {
		return nil, err
	}

	projectId, err := c.GetProjectId()
	if err != nil {
		return nil, err
	}

	serverConfig, err := svc.Projects.Zones.GetServerconfig(projectId, zone).Context(context.Background()).Do()
	if err != nil {
		return nil, err
	}

	c.log.Info("Getting server config succeeded")

	serverConfig.ValidMasterVersions = updateVersions(serverConfig.ValidMasterVersions)
	serverConfig.ValidNodeVersions = updateVersions(serverConfig.ValidNodeVersions)

	return serverConfig, nil
}

func updateVersions(validVersions []string) []string {
	log.Info("append `major.minor` K8S version format to valid GKE versions")

	var updatedVersions []string

	for _, v := range validVersions {
		version := strings.Split(v, ".")

		if len(version) >= 2 {
			majorMinor := fmt.Sprintf("%s.%s", version[0], version[1])
			if !utils.Contains(updatedVersions, majorMinor) && majorMinor != v {
				updatedVersions = append(updatedVersions, majorMinor, v)
			} else if !utils.Contains(updatedVersions, v) {
				updatedVersions = append(updatedVersions, v)
			}
		} else if !utils.Contains(updatedVersions, v) {
			updatedVersions = append(updatedVersions, v)
		}
	}

	return updatedVersions
}

// GetAllMachineTypesByZone lists supported machine types by zone
func (c *GKECluster) GetAllMachineTypesByZone(zone string) (map[string]pkgCluster.MachineTypes, error) {
	computeService, err := c.getComputeService()
	if err != nil {
		return nil, err
	}

	project, err := c.GetProjectId()
	if err != nil {
		return nil, err
	}

	return getMachineTypes(computeService, project, zone)
}

// GetAllMachineTypes lists all supported machine types
func (c *GKECluster) GetAllMachineTypes() (map[string]pkgCluster.MachineTypes, error) {
	computeService, err := c.getComputeService()
	if err != nil {
		return nil, err
	}

	project, err := c.GetProjectId()
	if err != nil {
		return nil, err
	}
	return getMachineTypesWithoutZones(computeService, project)
}

// getMachineTypesWithoutZones lists supported machine types in all zone
func getMachineTypesWithoutZones(csv *gkeCompute.Service, project string) (map[string]pkgCluster.MachineTypes, error) {
	response := make(map[string]pkgCluster.MachineTypes)
	req := csv.MachineTypes.AggregatedList(project)
	if err := req.Pages(context.Background(), func(list *gkeCompute.MachineTypeAggregatedList) error {
		for zone, item := range list.Items {
			var types []string
			for _, t := range item.MachineTypes {
				types = append(types, t.Name)
			}
			key := zone
			if strings.HasPrefix(key, zonePrefix) {
				key = zone[len(zonePrefix):]
			}
			if types != nil {
				response[key] = types
			}
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return response, nil
}

const zonePrefix = "zones/"

// getMachineTypes returns supported machine types by zone
func getMachineTypes(csv *gkeCompute.Service, project, zone string) (map[string]pkgCluster.MachineTypes, error) {
	var machineTypes []string
	req := csv.MachineTypes.List(project, zone)
	if err := req.Pages(context.Background(), func(page *gkeCompute.MachineTypeList) error {
		for _, machineType := range page.Items {
			machineTypes = append(machineTypes, machineType.Name)
		}
		return nil
	}); err != nil {
		return nil, err
	}

	response := make(map[string]pkgCluster.MachineTypes)
	response[zone] = machineTypes
	return response, nil
}

// getComputeService create a Compute Service from GKECluster
func (c *GKECluster) getComputeService() (*gkeCompute.Service, error) {
	// New client from credentials
	client, err := c.newClientFromCredentials()
	if err != nil {
		return nil, errors.WrapIf(err, "creating http client failed")
	}
	service, err := gkeCompute.New(client)
	if err != nil {
		return nil, errors.WrapIf(err, "instantiating google compute service failed")
	}
	return service, nil
}

// newClientFromCredentials creates new client from credentials
func (c *GKECluster) newClientFromCredentials() (*http.Client, error) {
	// Get Secret from Vault
	clusterSecret, err := c.GetSecretWithValidation()
	if err != nil {
		return nil, errors.WrapIf(err, "retrieving cluster credentials secret failed")
	}

	// TODO https://github.com/mitchellh/mapstructure

	credentials := pkgProviderGoogle.CreateServiceAccount(clusterSecret.Values)
	return pkgProviderGoogle.CreateOath2Client(credentials)
}

// GetZones lists supported zones
func (c *GKECluster) GetZones() ([]string, error) {
	computeService, err := c.getComputeService()
	if err != nil {
		return nil, err
	}

	project, err := c.GetProjectId()
	if err != nil {
		return nil, err
	}
	var zones []string
	req := computeService.Zones.List(project)
	if err := req.Pages(context.Background(), func(page *gkeCompute.ZoneList) error {
		for _, zone := range page.Items {
			zones = append(zones, zone.Name)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return zones, nil
}

// GetProjectId returns with project id from secret
func (c *GKECluster) GetProjectId() (string, error) {
	s, err := c.GetSecretWithValidation()
	if err != nil {
		return "", err
	}

	return s.Values[secrettype.ProjectId], nil
}

// SetStatus sets the cluster's status
func (c *GKECluster) SetStatus(status, statusMessage string) error {
	if c.model.Cluster.Status == status && c.model.Cluster.StatusMessage == statusMessage {
		return nil
	}

	if c.model.Cluster.ID != 0 {
		// Record status change to history before modifying the actual status.
		// If setting/saving the actual status doesn't succeed somehow, at least we can reconstruct it from history (i.e. event sourcing).
		statusHistory := clustermodel.StatusHistoryModel{
			ClusterID:   c.model.Cluster.ID,
			ClusterName: c.model.Cluster.Name,

			FromStatus:        c.model.Cluster.Status,
			FromStatusMessage: c.model.Cluster.StatusMessage,
			ToStatus:          status,
			ToStatusMessage:   statusMessage,
		}

		if err := c.repository.SaveStatusHistory(&statusHistory); err != nil {
			return errors.Wrap(err, "failed to record cluster status change to history")
		}
	}

	if c.model.Cluster.Status == pkgCluster.Creating && (status == pkgCluster.Running || status == pkgCluster.Warning) {
		now := time.Now()
		c.model.Cluster.StartedAt = &now
	}
	c.model.Cluster.Status = status
	c.model.Cluster.StatusMessage = statusMessage

	if err := c.repository.SaveModel(c.model); err != nil {
		return errors.Wrap(err, "failed to update cluster status")
	}

	return nil
}

// NodePoolExists returns true if node pool with nodePoolName exists
func (c *GKECluster) NodePoolExists(nodePoolName string) bool {
	for _, np := range c.model.NodePools {
		if np != nil && np.Name == nodePoolName {
			return true
		}
	}
	return false
}

func (c *GKECluster) setNodePoolSize(nodePoolName string, count int) bool {
	for _, np := range c.model.NodePools {
		if np != nil && np.Name == nodePoolName {
			np.NodeCount = count
		}
	}
	return false
}

// IsReady checks if the cluster is running according to the cloud provider.
func (c *GKECluster) IsReady() (bool, error) {
	c.log.Debug("Get Google Service Client")
	svc, err := c.getGoogleServiceClient()
	if err != nil {
		be := getBanzaiErrorFromError(err)
		return false, errors.New(be.Message)
	}
	c.log.Debug("Get Google Service Client success")

	secretItem, err := c.GetSecretWithValidation()
	if err != nil {
		return false, err
	}

	c.log.Debug("Get gke cluster")
	cl, err := svc.Projects.Zones.Clusters.Get(secretItem.Values[secrettype.ProjectId], c.model.Cluster.Location, c.model.Cluster.Name).Context(context.Background()).Do()
	if err != nil {
		apiError := getBanzaiErrorFromError(err)
		return false, errors.New(apiError.Message)
	}
	c.log.Debug("Get cluster success")

	return statusRunning == cl.Status, nil
}

// ValidateCreationFields validates all field
func (c *GKECluster) ValidateCreationFields(r *pkgCluster.CreateClusterRequest) error {
	location := r.Location

	// Validate location
	c.log.Info("Validate location")
	if err := c.validateLocation(location); err != nil {
		return err
	}
	c.log.Info("Validate location passed")

	// Validate machine types
	nodePools := r.Properties.CreateClusterGKE.NodePools
	c.log.Info("Validate nodePools")
	if err := c.validateMachineType(nodePools, location); err != nil {
		return err
	}
	c.log.Info("Validate nodePools passed")

	// Validate kubernetes version
	c.log.Info("Validate kubernetesVersion")
	masterVersion := r.Properties.CreateClusterGKE.Master.Version
	nodeVersion := r.Properties.CreateClusterGKE.NodeVersion
	if err := c.validateKubernetesVersion(masterVersion, nodeVersion, location); err != nil {
		return err
	}
	c.log.Info("Validate kubernetesVersion passed")

	// Validate vpc and subnet
	if r.Properties.CreateClusterGKE.Vpc != "" {
		log.Info("Validate VPC and subnet names")
		if err := c.validateVPCAndSubnet(r.Properties.CreateClusterGKE.Vpc, r.Properties.CreateClusterGKE.Subnet); err != nil {
			return err
		}
		log.Info("Validate VPC and subnet names passed")
	}

	return nil
}

func (c *GKECluster) validateVPCAndSubnet(VPCName string, subnetName string) error {
	project, err := c.GetProjectId()
	if err != nil {
		return errors.WrapIf(err, "could not get project id")
	}

	svc, err := c.getComputeService()
	if err != nil {
		return errors.WrapIf(err, "could not get compute service")
	}

	network, err := svc.Networks.Get(project, VPCName).Context(context.Background()).Do()
	if err != nil {
		if googleErr, ok := errors.Cause(err).(*googleapi.Error); ok {
			if googleErr.Code == http.StatusNotFound {
				return errors.WithDetails(errors.New("VPC not found"), "vpc", VPCName)
			}
		}
		return errors.WrapIfWithDetails(err, "could not get VPC", "vpc", VPCName)
	}

	if subnetName == "" {
		return nil
	}

	zone := c.GetLocation()
	region, err := findRegionByZone(svc, project, zone)
	if err != nil {
		return errors.WrapIfWithDetails(err, "could not find region by zone", "project", project, "zone", zone)
	}

	subnet, err := svc.Subnetworks.Get(project, region.Name, subnetName).Context(context.Background()).Do()
	if err != nil {
		if googleErr, ok := errors.Cause(err).(*googleapi.Error); ok {
			if googleErr.Code == http.StatusNotFound {
				return errors.WithDetails(errors.New("subnet not found"), "project", project, "subnet", subnetName, "region", region.Name)
			}
		}
		return err
	}

	if network.SelfLink != subnet.Network {
		return errors.WithDetails(errors.New("subnet is in a different VPC"), "project", project, "vpc", network.Name, "subnet", subnet.Name)
	}

	return nil
}

// validateLocation validates location
func (c *GKECluster) validateLocation(location string) error {
	c.log.Infof("Location: %s", location)
	validLocations, err := c.GetZones()
	if err != nil {
		return err
	}

	c.log.Infof("Valid locations: %v", validLocations)

	if isOk := utils.Contains(validLocations, location); !isOk {
		return pkgErrors.ErrorNotValidLocation
	}

	return nil
}

// validateMachineType validates nodeInstanceTypes
func (c *GKECluster) validateMachineType(nodePools map[string]*pkgClusterGoogle.NodePool, location string) error {
	var machineTypes []string
	for _, nodePool := range nodePools {
		if nodePool != nil {
			machineTypes = append(machineTypes, nodePool.NodeInstanceType)
		}
	}

	c.log.Infof("NodeInstanceTypes: %v", machineTypes)

	validMachineTypes, err := c.GetAllMachineTypesByZone(location)
	if err != nil {
		return err
	}
	c.log.Infof("Valid NodeInstanceTypes: %v", validMachineTypes[location])

	for _, mt := range machineTypes {
		if isOk := utils.Contains(validMachineTypes[location], mt); !isOk {
			return pkgErrors.ErrorNotValidNodeInstanceType
		}
	}

	return nil
}

// validateKubernetesVersion validates k8s versions
func (c *GKECluster) validateKubernetesVersion(masterVersion, nodeVersion, location string) error {
	c.log.Infof("Master version: %s", masterVersion)
	c.log.Infof("Node version: %s", nodeVersion)
	config, err := c.GetGkeServerConfig(location)
	if err != nil {
		return err
	}

	validNodeVersions := config.ValidNodeVersions
	log.Infof("Valid node versions: %s", validNodeVersions)

	if isOk := utils.Contains(validNodeVersions, nodeVersion); !isOk {
		return pkgErrors.ErrorNotValidNodeVersion
	}

	validMasterVersions := config.ValidMasterVersions
	c.log.Infof("Valid master versions: %s", validMasterVersions)

	if isOk := utils.Contains(validMasterVersions, masterVersion); !isOk {
		return pkgErrors.ErrorNotValidMasterVersion
	}

	return nil
}

// GetSecretWithValidation returns secret from vault
func (c *GKECluster) GetSecretWithValidation() (*secret.SecretItemResponse, error) {
	return c.CommonClusterBase.getSecret(c)
}

// SaveConfigSecretId saves the config secret id in database
func (c *GKECluster) SaveConfigSecretId(configSecretId string) error {
	c.model.Cluster.ConfigSecretID = configSecretId

	err := c.repository.SaveModel(c.model)
	if err != nil {
		return errors.Wrap(err, "failed to save config secret id")
	}

	return nil
}

// GetConfigSecretId return config secret id
func (c *GKECluster) GetConfigSecretId() string {
	return c.model.Cluster.ConfigSecretID
}

// GetK8sConfig returns the Kubernetes config
func (c *GKECluster) GetK8sConfig() ([]byte, error) {
	return c.CommonClusterBase.getConfig(c)
}

// GetK8sUserConfig returns the Kubernetes config
func (c *GKECluster) GetK8sUserConfig() ([]byte, error) {
	return c.GetK8sConfig()
}

func waitForOperation(getter operationInfoer, operationName string, logger logrus.FieldLogger) error {
	log := logger.WithFields(logrus.Fields{"operation": operationName})

	log.Info("start checking operation status")

	var operationType string
	var err error
	operationStatus := statusRunning
	for operationStatus != statusDone {
		operationStatus, operationType, err = getter.getInfo(operationName)
		if err != nil {
			return errors.WrapIf(err, "retrieving operation status failed")
		}

		log.Infof("Operation[%s] status: %s", operationType, operationStatus)
		time.Sleep(time.Second * 5)
	}

	return nil
}

// RbacEnabled returns true if rbac enabled on the cluster
func (c *GKECluster) RbacEnabled() bool {
	return c.model.Cluster.RbacEnabled
}

func hasTag(cluster *gke.Cluster, tagKey, tagValue string) bool {
	if l, ok := cluster.ResourceLabels[tagKey]; ok && l == tagValue {
		return true
	}
	return false
}
