/*
Copyright 2017 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 framework

import (
	"fmt"
	"time"

	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	restclient "k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
	"k8s.io/kubernetes/pkg/api/v1"
	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
	"k8s.io/kubernetes/test/e2e/framework"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
)

const federatedClustersWaitTimeout = 1 * time.Minute

// ClusterSlice is a slice of clusters
type ClusterSlice []*Cluster

// Cluster keeps track of the name and client of a cluster in the federation
type Cluster struct {
	Name string
	*kubeclientset.Clientset
}

// registeredClustersFromConfig configures clientsets for registered clusters from the e2e kubeconfig
func registeredClustersFromConfig(f *Framework) ClusterSlice {
	contexts := f.GetUnderlyingFederatedContexts()

	By("Obtaining a list of all the clusters")
	clusterList := waitForAllRegisteredClusters(f, len(contexts))

	framework.Logf("Checking that %d clusters are Ready", len(contexts))
	for _, context := range contexts {
		ClusterIsReadyOrFail(f, context.Name)

	}
	framework.Logf("%d clusters are Ready", len(contexts))

	clusters := ClusterSlice{}
	for i, c := range clusterList.Items {
		framework.Logf("Creating a clientset for the cluster %s", c.Name)
		Expect(framework.TestContext.KubeConfig).ToNot(Equal(""), "KubeConfig must be specified to load clusters' client config")
		config := restConfigFromContext(c, i)
		clusters = append(clusters, &Cluster{
			Name:      c.Name,
			Clientset: clientsetFromConfig(f, config, c.Spec.ServerAddressByClientCIDRs[0].ServerAddress),
		})

	}
	waitForNamespaceInFederatedClusters(clusters, f.FederationNamespace.Name)
	return clusters
}

// waitForAllRegisteredClusters waits for all clusters defined in e2e context to be created
// return ClusterList until the listed cluster items equals clusterCount
func waitForAllRegisteredClusters(f *Framework, clusterCount int) *federationapi.ClusterList {
	var clusterList *federationapi.ClusterList
	if err := wait.PollImmediate(framework.Poll, federatedClustersWaitTimeout, func() (bool, error) {
		var err error
		clusterList, err = f.FederationClientset.Federation().Clusters().List(metav1.ListOptions{})
		if err != nil {
			return false, err
		}
		framework.Logf("%d clusters registered, waiting for %d", len(clusterList.Items), clusterCount)
		if len(clusterList.Items) == clusterCount {
			return true, nil
		}
		return false, nil
	}); err != nil {
		framework.Failf("Failed to list registered clusters: %+v", err)
	}
	return clusterList
}

func restConfigFromContext(c federationapi.Cluster, i int) *restclient.Config {
	kubecfg, err := clientcmd.LoadFromFile(framework.TestContext.KubeConfig)
	framework.ExpectNoError(err, "error loading KubeConfig: %v", err)

	ccfg := clientcmd.NewNonInteractiveClientConfig(*kubecfg, c.Name, &clientcmd.ConfigOverrides{}, clientcmd.NewDefaultClientConfigLoadingRules())
	cfg, err := ccfg.ClientConfig()
	framework.ExpectNoError(err, "Error creating client config in cluster #%d (%q)", i, c.Name)
	return cfg
}

func clientsetFromConfig(f *Framework, cfg *restclient.Config, host string) *kubeclientset.Clientset {
	cfg.Host = host
	cfg.QPS = f.Framework.Options.ClientQPS
	cfg.Burst = f.Framework.Options.ClientBurst
	return kubeclientset.NewForConfigOrDie(restclient.AddUserAgent(cfg, "federation-e2e"))
}

// waitForNamespaceInFederatedClusters waits for the federated namespace to be created in federated clusters
func waitForNamespaceInFederatedClusters(clusters ClusterSlice, nsName string) {
	for _, c := range clusters {
		name := c.Name
		By(fmt.Sprintf("Waiting for namespace %q to be created in cluster %q", nsName, name))
		err := wait.PollImmediate(framework.Poll, FederatedDefaultTestTimeout, func() (bool, error) {
			_, err := c.Clientset.Core().Namespaces().Get(nsName, metav1.GetOptions{})
			if errors.IsNotFound(err) {
				return false, nil
			} else if err != nil {
				framework.Logf("An error occurred waiting for namespace %q to be created in cluster %q: %v", nsName, name, err)
				return false, nil
			}
			By(fmt.Sprintf("Namespace %q exists in cluster %q", nsName, name))
			return true, nil
		})
		framework.ExpectNoError(err, "Failed to verify federated namespace %q creation in cluster %q", nsName, name)
	}
}

// ClusterIsReadyOrFail checks whether the named cluster is ready
func ClusterIsReadyOrFail(f *Framework, clusterName string) {
	By(fmt.Sprintf("Checking readiness of cluster %q", clusterName))
	err := wait.PollImmediate(framework.Poll, FederatedDefaultTestTimeout, func() (bool, error) {
		c, err := f.FederationClientset.Federation().Clusters().Get(clusterName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, condition := range c.Status.Conditions {
			if condition.Type == federationapi.ClusterReady && condition.Status == v1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	})
	framework.ExpectNoError(err, fmt.Sprintf("Unexpected error in verifying if cluster %q is ready: %+v", clusterName, err))
	framework.Logf("Cluster %s is Ready", clusterName)
}