<commit_before>package slinga

import (
	"fmt"
	log "github.com/Sirupsen/logrus"
	"google.golang.org/grpc/grpclog"
	yaml "gopkg.in/yaml.v2"
	"k8s.io/helm/pkg/helm"
	"k8s.io/helm/pkg/helm/portforwarder"
	"k8s.io/helm/pkg/kube"
	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
	internalversioncore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
	"k8s.io/kubernetes/pkg/client/restclient"
	"k8s.io/kubernetes/pkg/labels"
	"strings"

	"errors"
	"github.com/mattn/go-zglob"
)

// HelmCodeExecutor is an executor that uses Helm for deployment of apps on kubernetes
type HelmCodeExecutor struct {
	Code     *Code
	Cluster  *Cluster
	Key      string
	Metadata map[string]string
	Params   NestedParameterMap
}

// NewHelmCodeExecutor constructs HelmCodeExecutor from given *Code
func NewHelmCodeExecutor(code *Code, key string, codeMetadata map[string]string, codeParams NestedParameterMap, clusters map[string]*Cluster) (CodeExecutor, error) {
	// First of all, redirect Helm/grpc logging to our own debug stream
	// We don't want these messages to be printed to Stdout/Stderr
	grpclog.SetLogger(debug)

	// todo: should we check key existence first?
	if clusterName, ok := codeParams["cluster"].(string); !ok {
		return nil, errors.New("Cluster param should be defined")
	} else if cluster, ok := clusters[clusterName]; ok {
		exec := HelmCodeExecutor{Code: code, Cluster: cluster, Key: key, Metadata: codeMetadata, Params: codeParams}
		err := exec.setupTillerConnection()
		if err != nil {
			return nil, err
		}
		return exec, nil
	} else {
		return nil, errors.New("Specified cluster is undefined")
	}
}

func (cluster *Cluster) newKubeClient() (*restclient.Config, *internalclientset.Clientset, error) {
	kubeContext := cluster.Metadata.KubeContext
	config, err := kube.GetConfig(kubeContext).ClientConfig()
	if err != nil {
		return nil, nil, fmt.Errorf("Could not get kubernetes config for context '%s': %s", kubeContext, err)
	}
	client, err := internalclientset.NewForConfig(config)
	if err != nil {
		return nil, nil, fmt.Errorf("Could not get kubernetes client: %s", err)
	}
	return config, client, nil
}

func (exec *HelmCodeExecutor) setupTillerConnection() error {
	if exec.Cluster.Metadata.tillerHost != "" {
		return nil
	}

	config, client, err := exec.Cluster.newKubeClient()
	if err != nil {
		return err
	}

	tillerNamespace := exec.Cluster.Metadata.TillerNamespace
	if tillerNamespace == "" {
		tillerNamespace = "kube-system"
	}
	tunnel, err := portforwarder.New(tillerNamespace, client, config)
	if err != nil {
		return err
	}

	exec.Cluster.Metadata.tillerHost = fmt.Sprintf("localhost:%d", tunnel.Local)

	debug.WithFields(log.Fields{
		"port": tunnel.Local,
	}).Info("Created k8s tunnel using local port")

	return nil
}

func (exec *HelmCodeExecutor) newHelmClient() *helm.Client {
	return helm.NewClient(helm.Host(exec.Cluster.Metadata.tillerHost))
}

func findHelmRelease(helmClient *helm.Client, name string) (bool, error) {
	resp, err := helmClient.ListReleases()
	if err != nil {
		return false, nil
	}

	for _, rel := range resp.Releases {
		if rel.Name == name {
			return true, nil
		}
	}

	return false, nil
}

func releaseName(key string) string {
	return strings.ToLower(EscapeName(key))
}

// Install for HelmCodeExecutor runs "helm install" for the corresponding helm chart
func (exec HelmCodeExecutor) Install() error {
	releaseName := releaseName(exec.Key)

	chartName, ok := exec.Metadata["chartName"]
	if !ok {
		return errors.New("Chart name is undefined")
	}

	helmClient := exec.newHelmClient()

	exists, err := findHelmRelease(helmClient, releaseName)
	if err != nil {
		debug.WithFields(log.Fields{
			"releaseName": releaseName,
			"error":       err,
		}).Fatal("Error while looking for release")
	}

	if exists {
		// If a release already exists, let's just go ahead and update it
		debug.WithFields(log.Fields{
			"releaseName": releaseName,
		}).Info("Found that release already exists. Calling an update on it")
		return exec.Update()
	}

	chartPath := getValidChartPath(chartName)

	vals, err := yaml.Marshal(exec.Params)
	if err != nil {
		return err
	}

	debug.WithFields(log.Fields{
		"release": releaseName,
		"chart":   chartName,
		"path":    chartPath,
		"params":  string(vals),
	}).Info("Installing Helm release")

	_, err = helmClient.InstallRelease(chartPath, "demo", helm.ReleaseName(releaseName), helm.ValueOverrides(vals), helm.InstallReuseName(true))
	if err != nil {
		return err
	}
	return nil
}

// Update for HelmCodeExecutor runs "helm update" for the corresponding helm chart
func (exec HelmCodeExecutor) Update() error {
	releaseName := releaseName(exec.Key)

	chartName := exec.Metadata["chartName"]

	helmClient := exec.newHelmClient()

	chartPath := getValidChartPath(chartName)

	vals, err := yaml.Marshal(exec.Params)
	if err != nil {
		return err
	}

	debug.WithFields(log.Fields{
		"release": releaseName,
		"chart":   chartName,
		"path":    chartPath,
		"params":  string(vals),
	}).Info("Updating Helm release")

	_, err = helmClient.UpdateRelease(releaseName, chartPath, helm.UpdateValueOverrides(vals))
	if err != nil {
		return err
	}

	return nil
}

func getValidChartPath(chartName string) string {
	files, _ := zglob.Glob(GetAptomiObjectFilePatternTgz(GetAptomiBaseDir(), TypeCharts, chartName))
	if len(files) <= 0 {
		debug.WithFields(log.Fields{
			"chartName": chartName,
		}).Fatal("No charts found")
	}
	if len(files) > 1 {
		debug.WithFields(log.Fields{
			"chartName": chartName,
			"files":     files,
		}).Fatal("More that one chart found")
	}
	return files[0]
}

// Destroy for HelmCodeExecutor runs "helm delete" for the corresponding helm chart
func (exec HelmCodeExecutor) Destroy() error {
	releaseName := releaseName(exec.Key)

	helmClient := exec.newHelmClient()

	debug.WithFields(log.Fields{
		"release": releaseName,
	}).Info("Deleting Helm release")

	if _, err := helmClient.DeleteRelease(releaseName, helm.DeletePurge(true)); err != nil {
		return err
	}

	return nil
}

// Endpoints returns map from port type to url for all services of the current chart
func (exec HelmCodeExecutor) Endpoints() (map[string]string, error) {
	_, clientset, err := exec.Cluster.newKubeClient()
	if err != nil {
		return nil, err
	}

	coreClient := clientset.Core()

	releaseName := releaseName(exec.Key)
	chartName := exec.Metadata["chartName"]

	selector := labels.Set{"release": releaseName, "chart": chartName}.AsSelector()
	options := api.ListOptions{LabelSelector: selector}

	endpoints := make(map[string]string)

	// Check all corresponding services
	services, err := coreClient.Services(exec.Cluster.Metadata.Namespace).List(options)
	if err != nil {
		return nil, err
	}

	kubeHost, err := exec.getKubeExternalAddress(coreClient)
	if err != nil {
		return nil, err
	}

	for _, service := range services.Items {
		if service.Spec.Type == "NodePort" {
			for _, port := range service.Spec.Ports {
				sURL := fmt.Sprintf("%s:%d", kubeHost, port.NodePort)

				// todo(slukjanov): could we somehow detect real schema? I think no :(
				if stringContainsAny(port.Name, "https") {
					sURL = "https://" + sURL
				} else if stringContainsAny(port.Name, "ui", "rest", "http", "grafana") {
					sURL = "http://" + sURL
				}

				endpoints[port.Name] = sURL
			}
		}
	}

	// Check all corresponding istio ingresses
	ingresses, err := clientset.Extensions().Ingresses(exec.Cluster.Metadata.Namespace).List(options)
	if err != nil {
		return nil, err
	}

	// Find Istio Ingress service (how ingress itself exposed)
	service, err := coreClient.Services(exec.Cluster.Metadata.Namespace).Get("istio-ingress")
	if err != nil {
		return nil, err
	}
	istioIngress := "<unresolved>"
	if service.Spec.Type == "NodePort" {
		for _, port := range service.Spec.Ports {
			if port.Name == "http" {
				istioIngress = fmt.Sprintf("%s:%d", kubeHost, port.NodePort)
			}
		}
	}

	// todo(slukjanov): support more then one ingress / rule / path
	for _, ingress := range ingresses.Items {
		if class, ok := ingress.Annotations["kubernetes.io/ingress.class"]; !ok || class != "istio" {
			continue
		}
		for _, rule := range ingress.Spec.Rules {
			for _, path := range rule.HTTP.Paths {
				pathStr := strings.Trim(path.Path, ".*")

				if rule.Host == "" {
					endpoints["ingress"] = "http://" + istioIngress + pathStr
				} else {
					endpoints["ingress"] = "http://" + rule.Host + pathStr
				}
			}
		}
	}

	return endpoints, nil
}

func (exec HelmCodeExecutor) getKubeExternalAddress(client internalversioncore.CoreInterface) (string, error) {
	if exec.Cluster.Metadata.kubeExternalAddress != "" {
		return exec.Cluster.Metadata.kubeExternalAddress, nil
	}
<commit_msg>Don't fail if can't find istio service - just ignore<commit_after>package slinga

import (
	"fmt"
	log "github.com/Sirupsen/logrus"
	"google.golang.org/grpc/grpclog"
	yaml "gopkg.in/yaml.v2"
	"k8s.io/helm/pkg/helm"
	"k8s.io/helm/pkg/helm/portforwarder"
	"k8s.io/helm/pkg/kube"
	"k8s.io/kubernetes/pkg/api"
	k8serrors "k8s.io/kubernetes/pkg/api/errors"
	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
	internalversioncore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
	"k8s.io/kubernetes/pkg/client/restclient"
	"k8s.io/kubernetes/pkg/labels"
	"strings"

	"errors"
	"github.com/mattn/go-zglob"
)

// HelmCodeExecutor is an executor that uses Helm for deployment of apps on kubernetes
type HelmCodeExecutor struct {
	Code     *Code
	Cluster  *Cluster
	Key      string
	Metadata map[string]string
	Params   NestedParameterMap
}

// NewHelmCodeExecutor constructs HelmCodeExecutor from given *Code
func NewHelmCodeExecutor(code *Code, key string, codeMetadata map[string]string, codeParams NestedParameterMap, clusters map[string]*Cluster) (CodeExecutor, error) {
	// First of all, redirect Helm/grpc logging to our own debug stream
	// We don't want these messages to be printed to Stdout/Stderr
	grpclog.SetLogger(debug)

	// todo: should we check key existence first?
	if clusterName, ok := codeParams["cluster"].(string); !ok {
		return nil, errors.New("Cluster param should be defined")
	} else if cluster, ok := clusters[clusterName]; ok {
		exec := HelmCodeExecutor{Code: code, Cluster: cluster, Key: key, Metadata: codeMetadata, Params: codeParams}
		err := exec.setupTillerConnection()
		if err != nil {
			return nil, err
		}
		return exec, nil
	} else {
		return nil, errors.New("Specified cluster is undefined")
	}
}

func (cluster *Cluster) newKubeClient() (*restclient.Config, *internalclientset.Clientset, error) {
	kubeContext := cluster.Metadata.KubeContext
	config, err := kube.GetConfig(kubeContext).ClientConfig()
	if err != nil {
		return nil, nil, fmt.Errorf("Could not get kubernetes config for context '%s': %s", kubeContext, err)
	}
	client, err := internalclientset.NewForConfig(config)
	if err != nil {
		return nil, nil, fmt.Errorf("Could not get kubernetes client: %s", err)
	}
	return config, client, nil
}

func (exec *HelmCodeExecutor) setupTillerConnection() error {
	if exec.Cluster.Metadata.tillerHost != "" {
		return nil
	}

	config, client, err := exec.Cluster.newKubeClient()
	if err != nil {
		return err
	}

	tillerNamespace := exec.Cluster.Metadata.TillerNamespace
	if tillerNamespace == "" {
		tillerNamespace = "kube-system"
	}
	tunnel, err := portforwarder.New(tillerNamespace, client, config)
	if err != nil {
		return err
	}

	exec.Cluster.Metadata.tillerHost = fmt.Sprintf("localhost:%d", tunnel.Local)

	debug.WithFields(log.Fields{
		"port": tunnel.Local,
	}).Info("Created k8s tunnel using local port")

	return nil
}

func (exec *HelmCodeExecutor) newHelmClient() *helm.Client {
	return helm.NewClient(helm.Host(exec.Cluster.Metadata.tillerHost))
}

func findHelmRelease(helmClient *helm.Client, name string) (bool, error) {
	resp, err := helmClient.ListReleases()
	if err != nil {
		return false, nil
	}

	for _, rel := range resp.Releases {
		if rel.Name == name {
			return true, nil
		}
	}

	return false, nil
}

func releaseName(key string) string {
	return strings.ToLower(EscapeName(key))
}

// Install for HelmCodeExecutor runs "helm install" for the corresponding helm chart
func (exec HelmCodeExecutor) Install() error {
	releaseName := releaseName(exec.Key)

	chartName, ok := exec.Metadata["chartName"]
	if !ok {
		return errors.New("Chart name is undefined")
	}

	helmClient := exec.newHelmClient()

	exists, err := findHelmRelease(helmClient, releaseName)
	if err != nil {
		debug.WithFields(log.Fields{
			"releaseName": releaseName,
			"error":       err,
		}).Fatal("Error while looking for release")
	}

	if exists {
		// If a release already exists, let's just go ahead and update it
		debug.WithFields(log.Fields{
			"releaseName": releaseName,
		}).Info("Found that release already exists. Calling an update on it")
		return exec.Update()
	}

	chartPath := getValidChartPath(chartName)

	vals, err := yaml.Marshal(exec.Params)
	if err != nil {
		return err
	}

	debug.WithFields(log.Fields{
		"release": releaseName,
		"chart":   chartName,
		"path":    chartPath,
		"params":  string(vals),
	}).Info("Installing Helm release")

	_, err = helmClient.InstallRelease(chartPath, "demo", helm.ReleaseName(releaseName), helm.ValueOverrides(vals), helm.InstallReuseName(true))
	if err != nil {
		return err
	}
	return nil
}

// Update for HelmCodeExecutor runs "helm update" for the corresponding helm chart
func (exec HelmCodeExecutor) Update() error {
	releaseName := releaseName(exec.Key)

	chartName := exec.Metadata["chartName"]

	helmClient := exec.newHelmClient()

	chartPath := getValidChartPath(chartName)

	vals, err := yaml.Marshal(exec.Params)
	if err != nil {
		return err
	}

	debug.WithFields(log.Fields{
		"release": releaseName,
		"chart":   chartName,
		"path":    chartPath,
		"params":  string(vals),
	}).Info("Updating Helm release")

	_, err = helmClient.UpdateRelease(releaseName, chartPath, helm.UpdateValueOverrides(vals))
	if err != nil {
		return err
	}

	return nil
}

func getValidChartPath(chartName string) string {
	files, _ := zglob.Glob(GetAptomiObjectFilePatternTgz(GetAptomiBaseDir(), TypeCharts, chartName))
	if len(files) <= 0 {
		debug.WithFields(log.Fields{
			"chartName": chartName,
		}).Fatal("No charts found")
	}
	if len(files) > 1 {
		debug.WithFields(log.Fields{
			"chartName": chartName,
			"files":     files,
		}).Fatal("More that one chart found")
	}
	return files[0]
}

// Destroy for HelmCodeExecutor runs "helm delete" for the corresponding helm chart
func (exec HelmCodeExecutor) Destroy() error {
	releaseName := releaseName(exec.Key)

	helmClient := exec.newHelmClient()

	debug.WithFields(log.Fields{
		"release": releaseName,
	}).Info("Deleting Helm release")

	if _, err := helmClient.DeleteRelease(releaseName, helm.DeletePurge(true)); err != nil {
		return err
	}

	return nil
}

// Endpoints returns map from port type to url for all services of the current chart
func (exec HelmCodeExecutor) Endpoints() (map[string]string, error) {
	_, clientset, err := exec.Cluster.newKubeClient()
	if err != nil {
		return nil, err
	}

	coreClient := clientset.Core()

	releaseName := releaseName(exec.Key)
	chartName := exec.Metadata["chartName"]

	selector := labels.Set{"release": releaseName, "chart": chartName}.AsSelector()
	options := api.ListOptions{LabelSelector: selector}

	endpoints := make(map[string]string)

	// Check all corresponding services
	services, err := coreClient.Services(exec.Cluster.Metadata.Namespace).List(options)
	if err != nil {
		return nil, err
	}

	kubeHost, err := exec.getKubeExternalAddress(coreClient)
	if err != nil {
		return nil, err
	}

	for _, service := range services.Items {
		if service.Spec.Type == "NodePort" {
			for _, port := range service.Spec.Ports {
				sURL := fmt.Sprintf("%s:%d", kubeHost, port.NodePort)

				// todo(slukjanov): could we somehow detect real schema? I think no :(
				if stringContainsAny(port.Name, "https") {
					sURL = "https://" + sURL
				} else if stringContainsAny(port.Name, "ui", "rest", "http", "grafana") {
					sURL = "http://" + sURL
				}

				endpoints[port.Name] = sURL
			}
		}
	}

	// Find Istio Ingress service (how ingress itself exposed)
	service, err := coreClient.Services(exec.Cluster.Metadata.Namespace).Get("istio-ingress")
	if err != nil {
		// return if there is no Istio deployed
		if k8serrors.IsNotFound(err) {
			return endpoints, nil
		}
		return nil, err
	}

	istioIngress := "<unresolved>"
	if service.Spec.Type == "NodePort" {
		for _, port := range service.Spec.Ports {
			if port.Name == "http" {
				istioIngress = fmt.Sprintf("%s:%d", kubeHost, port.NodePort)
			}
		}
	}

	// Check all corresponding istio ingresses
	ingresses, err := clientset.Extensions().Ingresses(exec.Cluster.Metadata.Namespace).List(options)
	if err != nil {
		return nil, err
	}

	// todo(slukjanov): support more then one ingress / rule / path
	for _, ingress := range ingresses.Items {
		if class, ok := ingress.Annotations["kubernetes.io/ingress.class"]; !ok || class != "istio" {
			continue
		}
		for _, rule := range ingress.Spec.Rules {
			for _, path := range rule.HTTP.Paths {
				pathStr := strings.Trim(path.Path, ".*")

				if rule.Host == "" {
					endpoints["ingress"] = "http://" + istioIngress + pathStr
				} else {
					endpoints["ingress"] = "http://" + rule.Host + pathStr
				}
			}
		}
	}

	return endpoints, nil
}

func (exec HelmCodeExecutor) getKubeExternalAddress(client internalversioncore.CoreInterface) (string, error) {
	if exec.Cluster.Metadata.kubeExternalAddress != "" {
		return exec.Cluster.Metadata.kubeExternalAddress, nil
	}

{'fraction_non_alphanumeric': 0.130877034358047, 'fraction_numerical': 0.0016387884267631103, 'mean_word_length': 4.221894364119209, 'pattern_counts': {'":': 36, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, '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': '3914064', 'n_tokens_mistral': 6228, 'n_tokens_neox': 5814, 'n_words': 1480}