<commit_before>package clusters

import (
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	oclient "github.com/openshift/origin/pkg/client"
	deployapi "github.com/openshift/origin/pkg/deploy/api"
	ocon "github.com/radanalyticsio/oshinko-cli/core/clusters/containers"
	odc "github.com/radanalyticsio/oshinko-cli/core/clusters/deploymentconfigs"
	opt "github.com/radanalyticsio/oshinko-cli/core/clusters/podtemplates"
	"github.com/radanalyticsio/oshinko-cli/core/clusters/probes"
	ort "github.com/radanalyticsio/oshinko-cli/core/clusters/routes"
	osv "github.com/radanalyticsio/oshinko-cli/core/clusters/services"
	kapi "k8s.io/kubernetes/pkg/api"
	kclient "k8s.io/kubernetes/pkg/client/unversioned"
	"k8s.io/kubernetes/pkg/labels"
	"k8s.io/kubernetes/pkg/selection"
	"k8s.io/kubernetes/pkg/util/sets"
)


const clusterConfigMsg = "invalid cluster configuration"
const missingConfigMsg = "unable to find spark configuration '%s'"
const findDepConfigMsg = "unable to find deployment configs"
const createDepConfigMsg = "unable to create deployment config '%s'"
const depMsg = "unable to find deployment config for spark %s"
const masterSrvMsg = "unable to create spark master service endpoint"
const masterWebSrvMsg = "unable to create spark master web service endpoint"
const mastermsg = "unable to find spark masters"
const updateDepMsg = "unable to update deployment config for spark %s"
const noSuchClusterMsg = "no such cluster '%s'"
const noClusterForDriverMsg = "no cluster found for app '%s'"
const ephemeralDelMsg = "cluster not deleted '%s'"
const podListMsg = "unable to retrive pod list"
const sparkImageMsg = "no spark image specified"
const clusterExistsMsg = "cluster '%s' already exists%s"

const typeLabel = "oshinko-type"
const clusterLabel = "oshinko-cluster"
const driverLabel = "uses-oshinko-cluster"
const ephemeralLabel = "ephemeral"

const workerType = "worker"
const masterType = "master"
const webuiType = "webui"

const masterPortName = "spark-master"
const masterPort = 7077
const webPortName = "spark-webui"
const webPort = 8080

const sparkconfdir = "/etc/oshinko-spark-configs"

// The suffix to add to the spark master hostname (clustername) for the web service
const webServiceSuffix = "-ui"

type SparkPod struct {
	IP     string
	Status string
	Type   string
}

type SparkCluster struct {
	Namespace    string `json:"namespace,omitempty"`
	Name         string `json:"name,omitempty"`
	Href         string `json:"href"`
	Image        string `json:"image"`
	MasterURL    string `json:"masterUrl"`
	MasterWebURL string `json:"masterWebUrl"`
	MasterWebRoute string `json:"masterWebRoute"`
	Status       string `json:"status"`
	WorkerCount  int    `json:"workerCount"`
	MasterCount  int    `json:"masterCount"`
	Config       ClusterConfig
	Ephemeral    string `json:"ephemeral,omitempty"`
	Pods         []SparkPod
}

func generalErr(err error, msg string, code int) ClusterError {
	if err != nil {
		if msg == "" {
			msg = "error: " + err.Error()
		} else {
			msg = msg + ", error: " + err.Error()
		}
	}
	return NewClusterError(msg, code)
}

func makeSelector(otype string, clustername string) kapi.ListOptions {
	// Build a selector list based on type and/or cluster name
	var ot *labels.Requirement
	var cname *labels.Requirement
	ls := labels.NewSelector()
	if otype == "" {
		ot, _ = labels.NewRequirement(typeLabel, selection.Exists, sets.String{})
	} else {
		ot, _ = labels.NewRequirement(typeLabel, selection.Equals, sets.NewString(otype))
	}
	ls = ls.Add(*ot)
	if clustername == "" {
		cname, _ = labels.NewRequirement(clusterLabel, selection.Exists, sets.String{})
	} else {
		cname, _ = labels.NewRequirement(clusterLabel, selection.Equals, sets.NewString(clustername))
	}
	ls = ls.Add(*cname)
	return kapi.ListOptions{LabelSelector: ls}
}

func retrieveServiceURL(client kclient.ServiceInterface, stype, clustername string) string {
	selectorlist := makeSelector(stype, clustername)
	srvs, err := client.List(selectorlist)
	if err == nil && len(srvs.Items) != 0 {
		srv := srvs.Items[0]
		scheme := "http://"
		if stype == masterType {
			scheme = "spark://"
		}
		return scheme + srv.Name + ":" + strconv.Itoa(int(srv.Spec.Ports[0].Port))
	}
	return "<missing>"
}

func retrieveRouteForService(client oclient.RouteInterface, stype, clustername string) string {
	selectorlist := makeSelector(stype, clustername)
	routes, err := client.List(selectorlist)
	if err == nil && len(routes.Items) != 0 {
		route := routes.Items[0]
		return route.Spec.Host
	}
	return ""
}

func checkForDeploymentConfigs(client oclient.DeploymentConfigInterface, clustername string) (bool, *deployapi.DeploymentConfig, error) {
	selectorlist := makeSelector(masterType, clustername)
	dcs, err := client.List(selectorlist)
	if err != nil {
		return false, nil, err
	}
	if len(dcs.Items) == 0 {
		return false, nil, nil
	}
	m := dcs.Items[0]
	selectorlist = makeSelector(workerType, clustername)
	dcs, err = client.List(selectorlist)
	if err != nil {
		return false, &m, err
	}
	if len(dcs.Items) == 0 {
		return false, &m, nil
	}
	return true, &m, nil

}

func makeEnvVars(clustername, sparkconfdir string) []kapi.EnvVar {
	envs := []kapi.EnvVar{}

	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_SPARK_CLUSTER", Value: clustername})
	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_REST_HOST", Value: os.Getenv("OSHINKO_REST_SERVICE_HOST")})
	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_REST_PORT", Value: os.Getenv("OSHINKO_REST_SERVICE_PORT")})
	if sparkconfdir != "" {
		envs = append(envs, kapi.EnvVar{Name: "UPDATE_SPARK_CONF_DIR", Value: sparkconfdir})
	}

	return envs
}

func makeWorkerEnvVars(clustername, sparkconfdir string) []kapi.EnvVar {
	envs := []kapi.EnvVar{}

	envs = makeEnvVars(clustername, sparkconfdir)
	envs = append(envs, kapi.EnvVar{
		Name:  "SPARK_MASTER_ADDRESS",
		Value: "spark://" + clustername + ":" + strconv.Itoa(masterPort)})
	envs = append(envs, kapi.EnvVar{
		Name:  "SPARK_MASTER_UI_ADDRESS",
		Value: "http://" + clustername + webServiceSuffix + ":" + strconv.Itoa(webPort)})
	return envs
}

func sparkWorker(namespace, image string, replicas int, clustername, sparkconfdir, sparkworkerconfig string) *odc.ODeploymentConfig {

	// Create the basic deployment config
	// We will use a label and pod selector based on the cluster name.
	// Openshift will add additional labels and selectors to distinguish pods handled by
	// this deploymentconfig from pods beloning to another.
	dc := odc.DeploymentConfig(clustername+"-w", namespace).
		TriggerOnConfigChange().RollingStrategy().Label(clusterLabel, clustername).
		Label(typeLabel, workerType).
		PodSelector(clusterLabel, clustername).Replicas(replicas)

	// Create a pod template spec with the matching label
	pt := opt.PodTemplateSpec().Label(clusterLabel, clustername).Label(typeLabel, workerType)

	// Create a container with the correct ports and start command
	webport := 8081
	webp := ocon.ContainerPort(webPortName, webport)
	cont := ocon.Container(dc.Name, image).
		Ports(webp).
		SetLivenessProbe(probes.NewHTTPGetProbe(webport)).EnvVars(makeWorkerEnvVars(clustername, sparkconfdir))

	if sparkworkerconfig != "" {
		pt = pt.SetConfigMapVolume(sparkworkerconfig)
		cont = cont.SetVolumeMount(sparkworkerconfig, sparkconfdir, true)
	}

	// Finally, assign the container to the pod template spec and
	// assign the pod template spec to the deployment config
	return dc.PodTemplateSpec(pt.Containers(cont))
}

func mastername(clustername string) string {
	return clustername + "-m"
}

func workername(clustername string) string {
	return clustername + "-w"
}

func sparkMaster(namespace, image string, replicas int, clustername, sparkconfdir, sparkmasterconfig, driverdc string) *odc.ODeploymentConfig {

	// Create the basic deployment config
	// We will use a label and pod selector based on the cluster name
	// Openshift will add additional labels and selectors to distinguish pods handled by
	// this deploymentconfig from pods beloning to another.
	dc := odc.DeploymentConfig(mastername(clustername), namespace).
		TriggerOnConfigChange().RollingStrategy().Label(clusterLabel, clustername).
		Label(typeLabel, masterType).
		PodSelector(clusterLabel, clustername).Replicas(replicas)

	if driverdc != "" {
		dc = dc.Label(ephemeralLabel, driverdc)
	}

	// Create a pod template spec with the matching label
	pt := opt.PodTemplateSpec().Label(clusterLabel, clustername).
		Label(typeLabel, masterType)

	// Create a container with the correct ports and start command
	liveness := probes.NewExecProbe([]string{"/bin/bash", "-c", "curl localhost:8080 | grep -e Status.*ALIVE"})
	liveness.InitialDelaySeconds = 10
	readiness := probes.NewHTTPGetProbe(webPort)
	masterp := ocon.ContainerPort(masterPortName, masterPort)
	webp := ocon.ContainerPort(webPortName, webPort)
	cont := ocon.Container(dc.Name, image).
		Ports(masterp, webp).
		SetLivenessProbe(liveness).
		SetReadinessProbe(readiness).EnvVars(makeEnvVars(clustername, sparkconfdir))

	if sparkmasterconfig != "" {
		pt = pt.SetConfigMapVolume(sparkmasterconfig)
		cont = cont.SetVolumeMount(sparkmasterconfig, sparkconfdir, true)
	}

	// Finally, assign the container to the pod template spec and
	// assign the pod template spec to the deployment config
	return dc.PodTemplateSpec(pt.Containers(cont))
}

func service(name string,
	port int,
	clustername, otype string,
	podselectors map[string]string) (*osv.OService, *osv.OServicePort) {

	p := osv.ServicePort(port).TargetPort(port)
	return osv.Service(name).Label(clusterLabel, clustername).
		Label(typeLabel, otype).PodSelectors(podselectors).Ports(p), p
}

func checkForConfigMap(name string, cm kclient.ConfigMapsInterface) error {
	_, err := cm.Get(name)
	if err != nil {
		if strings.Index(err.Error(), "not found") != -1 {
			return generalErr(err, fmt.Sprintf(missingConfigMsg, name), ClusterConfigCode)
		}
		return generalErr(nil, fmt.Sprintf(missingConfigMsg, name), ClientOperationCode)
	}
	return nil
}

func countWorkers(client kclient.PodInterface, clustername string) (int, *kapi.PodList, error) {
	// If we are  unable to retrieve a list of worker pods, return -1 for count
	// This is an error case, differnt from a list of length 0. Let the caller
	// decide whether to report the error or the -1 count
	cnt := -1
	selectorlist := makeSelector(workerType, clustername)
	pods, err := client.List(selectorlist)
	if pods != nil {
		cnt = len(pods.Items)
	}
	return cnt, pods, err
}

func getDriverDeployment(app, namespace string, client *kclient.Client) string {

	// When we make calls from a driver pod, the most likely value we have is a deployment
	// so use that first
	rcc := client.ReplicationControllers(namespace)
	rc, err := rcc.Get(app)
	if err == nil && rc != nil {
		return app
	}

	// Okay, it wasn't a deployment, maybe it's a pod
	pc := client.Pods(namespace)
	pod, err := pc.Get(app)
	if err == nil && pod != nil {
		return pod.Labels["deployment"]
	}
	return ""
}

// Create a cluster and return the representation
func CreateCluster(
	clustername, namespace, sparkimage string,
	config *ClusterConfig, osclient *oclient.Client, client kclient.Interface, app string, ephemeral bool) (SparkCluster, error) {

	var driverrc string
	var ephem_val string
	var masterconfdir string
	var workerconfdir string
	var result SparkCluster = SparkCluster{}

	createCode := func(err error) int {
		if err != nil && strings.Index(err.Error(), "already exists") != -1 {
			return ComponentExistsCode
		}
		return ClientOperationCode
	}

	masterhost := clustername

	// Check to see if a cluster already exists of the same name (complete or incomplete)
	existing := SparkCluster{}
	findClusterBody(clustername, namespace, osclient, client, &existing)
	if !CheckNoCluster(&existing) {
		var msg string
		if existing.Status != "Running" {
			msg = fmt.Sprintf(clusterExistsMsg, clustername, " (incomplete)")
		} else {
			msg = fmt.Sprintf(clusterExistsMsg, clustername, "")
		}
		return result, generalErr(nil, msg, ComponentExistsCode)
	}


	// Copy any named config referenced and update it with any explicit config values
	finalconfig, err := GetClusterConfig(config, client.ConfigMaps(namespace))
	if err != nil {
		return result, generalErr(err, clusterConfigMsg, ErrorCode(err))
	}
	if finalconfig.SparkImage != "" {
		sparkimage = finalconfig.SparkImage
	} else if sparkimage == "" {
		return result, generalErr(nil, sparkImageMsg, ClusterConfigCode)
	}

	mastercount := int(finalconfig.MasterCount)
	workercount := int(finalconfig.WorkerCount)

	// Check if finalconfig contains the names of ConfigMaps to use for spark
	// configuration. If so they must exist. The ConfigMaps will be mounted
	// as volumes on spark pods and the path stored in the environment
	// variable UPDATE_SPARK_CONF_DIR
	cm := client.ConfigMaps(namespace)
	if finalconfig.SparkMasterConfig != "" {
		err := checkForConfigMap(finalconfig.SparkMasterConfig, cm)
		if err != nil {
			return result, err
		}
		masterconfdir = sparkconfdir
	}

	if finalconfig.SparkWorkerConfig != "" {
		err := checkForConfigMap(finalconfig.SparkWorkerConfig, cm)
		if err != nil {
			return result, err
		}
		workerconfdir = sparkconfdir
	}

	// If an app value was passed find the deployment so we can do the cluster
	// association and potentially mark the cluster as ephemeral
	if app != "" {
		driverrc = getDriverDeployment(app, namespace, client)
	}
	if ephemeral {
		// If we can't find an rc then we just make a long-running cluster
		ephem_val = driverrc
	}

	// Create the master deployment config
	masterdc := sparkMaster(namespace, sparkimage, mastercount, clustername, masterconfdir, finalconfig.SparkMasterConfig, ephem_val)

	// Create the services that will be associated with the master pod
	// They will be created with selectors based on the pod labels
	mastersv, _ := service(masterhost,
		masterdc.FindPort(masterPortName),
		clustername, masterType,
		masterdc.GetPodTemplateSpecLabels())

	websv, _ := service(masterhost+webServiceSuffix,
		masterdc.FindPort(webPortName),
		clustername, webuiType,
		masterdc.GetPodTemplateSpecLabels())

	webuiroute := ort.NewRoute(websv.GetName() + "-route", websv.GetName(), clustername, "webui")

	// Create the worker deployment config
	workerdc := sparkWorker(namespace, sparkimage, workercount, clustername, workerconfdir, finalconfig.SparkWorkerConfig)

	// Launch all of the objects
	dcc := osclient.DeploymentConfigs(namespace)
	_, err = dcc.Create(&masterdc.DeploymentConfig)
	if err != nil {
		return result, generalErr(err, fmt.Sprintf(createDepConfigMsg, masterdc.Name), createCode(err))
	}
	_, err = dcc.Create(&workerdc.DeploymentConfig)
	if err != nil {
		// Since we created the master deployment config, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, fmt.Sprintf(createDepConfigMsg, workerdc.Name), createCode(err))
	}

	sc := client.Services(namespace)
	_, err = sc.Create(&mastersv.Service)
	if err != nil {
		// Since we've created things, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, masterSrvMsg, createCode(err))
	}


	_, err = sc.Create(&websv.Service)
	if err != nil {
		// Since we've created things, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, masterWebSrvMsg, createCode(err))
	}

	// We will expose the Spark master webui unless we are told not to do it
	if config.ExposeWebUI {
		rc := osclient.Routes(namespace)
		_, err = rc.Create(webuiroute)
	}

	// Wait for the replication controllers to exist before building the response.
	rcc := client.ReplicationControllers(namespace)
	{
		var mrepl, wrepl *kapi.ReplicationController
		mrepl = nil
		wrepl = nil
		for i := 0; i < 4; i++ {
			if mrepl == nil {
				mrepl, _ = getReplController(rcc, clustername, masterType)
			}
			if wrepl == nil {
				wrepl, _ = getReplController(rcc, clustername, workerType)
			}
			if wrepl != nil && mrepl != nil {
				break
			}
			time.Sleep(250 * time.Millisecond)
		}
	}

	// Now that the creation actually worked, label the dc if the app value was passed
	if driverrc != "" {
		driver, err := rcc.Get(driverrc)
		if err == nil {
			if driver.Labels == nil {
				driver.Labels = map[string]string{}
			}
			driver.Labels[driverLabel] = clustername
			rcc.Update(driver)
		}
	}
<commit_msg>Retry update on driver rc during create<commit_after>package clusters

import (
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	oclient "github.com/openshift/origin/pkg/client"
	deployapi "github.com/openshift/origin/pkg/deploy/api"
	ocon "github.com/radanalyticsio/oshinko-cli/core/clusters/containers"
	odc "github.com/radanalyticsio/oshinko-cli/core/clusters/deploymentconfigs"
	opt "github.com/radanalyticsio/oshinko-cli/core/clusters/podtemplates"
	"github.com/radanalyticsio/oshinko-cli/core/clusters/probes"
	ort "github.com/radanalyticsio/oshinko-cli/core/clusters/routes"
	osv "github.com/radanalyticsio/oshinko-cli/core/clusters/services"
	kapi "k8s.io/kubernetes/pkg/api"
	kclient "k8s.io/kubernetes/pkg/client/unversioned"
	"k8s.io/kubernetes/pkg/labels"
	"k8s.io/kubernetes/pkg/selection"
	"k8s.io/kubernetes/pkg/util/sets"
)


const clusterConfigMsg = "invalid cluster configuration"
const missingConfigMsg = "unable to find spark configuration '%s'"
const findDepConfigMsg = "unable to find deployment configs"
const createDepConfigMsg = "unable to create deployment config '%s'"
const depMsg = "unable to find deployment config for spark %s"
const masterSrvMsg = "unable to create spark master service endpoint"
const masterWebSrvMsg = "unable to create spark master web service endpoint"
const mastermsg = "unable to find spark masters"
const updateDepMsg = "unable to update deployment config for spark %s"
const noSuchClusterMsg = "no such cluster '%s'"
const noClusterForDriverMsg = "no cluster found for app '%s'"
const ephemeralDelMsg = "cluster not deleted '%s'"
const podListMsg = "unable to retrive pod list"
const sparkImageMsg = "no spark image specified"
const clusterExistsMsg = "cluster '%s' already exists%s"

const typeLabel = "oshinko-type"
const clusterLabel = "oshinko-cluster"
const driverLabel = "uses-oshinko-cluster"
const ephemeralLabel = "ephemeral"

const workerType = "worker"
const masterType = "master"
const webuiType = "webui"

const masterPortName = "spark-master"
const masterPort = 7077
const webPortName = "spark-webui"
const webPort = 8080

const sparkconfdir = "/etc/oshinko-spark-configs"

// The suffix to add to the spark master hostname (clustername) for the web service
const webServiceSuffix = "-ui"

type SparkPod struct {
	IP     string
	Status string
	Type   string
}

type SparkCluster struct {
	Namespace    string `json:"namespace,omitempty"`
	Name         string `json:"name,omitempty"`
	Href         string `json:"href"`
	Image        string `json:"image"`
	MasterURL    string `json:"masterUrl"`
	MasterWebURL string `json:"masterWebUrl"`
	MasterWebRoute string `json:"masterWebRoute"`
	Status       string `json:"status"`
	WorkerCount  int    `json:"workerCount"`
	MasterCount  int    `json:"masterCount"`
	Config       ClusterConfig
	Ephemeral    string `json:"ephemeral,omitempty"`
	Pods         []SparkPod
}

func generalErr(err error, msg string, code int) ClusterError {
	if err != nil {
		if msg == "" {
			msg = "error: " + err.Error()
		} else {
			msg = msg + ", error: " + err.Error()
		}
	}
	return NewClusterError(msg, code)
}

func makeSelector(otype string, clustername string) kapi.ListOptions {
	// Build a selector list based on type and/or cluster name
	var ot *labels.Requirement
	var cname *labels.Requirement
	ls := labels.NewSelector()
	if otype == "" {
		ot, _ = labels.NewRequirement(typeLabel, selection.Exists, sets.String{})
	} else {
		ot, _ = labels.NewRequirement(typeLabel, selection.Equals, sets.NewString(otype))
	}
	ls = ls.Add(*ot)
	if clustername == "" {
		cname, _ = labels.NewRequirement(clusterLabel, selection.Exists, sets.String{})
	} else {
		cname, _ = labels.NewRequirement(clusterLabel, selection.Equals, sets.NewString(clustername))
	}
	ls = ls.Add(*cname)
	return kapi.ListOptions{LabelSelector: ls}
}

func retrieveServiceURL(client kclient.ServiceInterface, stype, clustername string) string {
	selectorlist := makeSelector(stype, clustername)
	srvs, err := client.List(selectorlist)
	if err == nil && len(srvs.Items) != 0 {
		srv := srvs.Items[0]
		scheme := "http://"
		if stype == masterType {
			scheme = "spark://"
		}
		return scheme + srv.Name + ":" + strconv.Itoa(int(srv.Spec.Ports[0].Port))
	}
	return "<missing>"
}

func retrieveRouteForService(client oclient.RouteInterface, stype, clustername string) string {
	selectorlist := makeSelector(stype, clustername)
	routes, err := client.List(selectorlist)
	if err == nil && len(routes.Items) != 0 {
		route := routes.Items[0]
		return route.Spec.Host
	}
	return ""
}

func checkForDeploymentConfigs(client oclient.DeploymentConfigInterface, clustername string) (bool, *deployapi.DeploymentConfig, error) {
	selectorlist := makeSelector(masterType, clustername)
	dcs, err := client.List(selectorlist)
	if err != nil {
		return false, nil, err
	}
	if len(dcs.Items) == 0 {
		return false, nil, nil
	}
	m := dcs.Items[0]
	selectorlist = makeSelector(workerType, clustername)
	dcs, err = client.List(selectorlist)
	if err != nil {
		return false, &m, err
	}
	if len(dcs.Items) == 0 {
		return false, &m, nil
	}
	return true, &m, nil

}

func makeEnvVars(clustername, sparkconfdir string) []kapi.EnvVar {
	envs := []kapi.EnvVar{}

	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_SPARK_CLUSTER", Value: clustername})
	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_REST_HOST", Value: os.Getenv("OSHINKO_REST_SERVICE_HOST")})
	envs = append(envs, kapi.EnvVar{Name: "OSHINKO_REST_PORT", Value: os.Getenv("OSHINKO_REST_SERVICE_PORT")})
	if sparkconfdir != "" {
		envs = append(envs, kapi.EnvVar{Name: "UPDATE_SPARK_CONF_DIR", Value: sparkconfdir})
	}

	return envs
}

func makeWorkerEnvVars(clustername, sparkconfdir string) []kapi.EnvVar {
	envs := []kapi.EnvVar{}

	envs = makeEnvVars(clustername, sparkconfdir)
	envs = append(envs, kapi.EnvVar{
		Name:  "SPARK_MASTER_ADDRESS",
		Value: "spark://" + clustername + ":" + strconv.Itoa(masterPort)})
	envs = append(envs, kapi.EnvVar{
		Name:  "SPARK_MASTER_UI_ADDRESS",
		Value: "http://" + clustername + webServiceSuffix + ":" + strconv.Itoa(webPort)})
	return envs
}

func sparkWorker(namespace, image string, replicas int, clustername, sparkconfdir, sparkworkerconfig string) *odc.ODeploymentConfig {

	// Create the basic deployment config
	// We will use a label and pod selector based on the cluster name.
	// Openshift will add additional labels and selectors to distinguish pods handled by
	// this deploymentconfig from pods beloning to another.
	dc := odc.DeploymentConfig(clustername+"-w", namespace).
		TriggerOnConfigChange().RollingStrategy().Label(clusterLabel, clustername).
		Label(typeLabel, workerType).
		PodSelector(clusterLabel, clustername).Replicas(replicas)

	// Create a pod template spec with the matching label
	pt := opt.PodTemplateSpec().Label(clusterLabel, clustername).Label(typeLabel, workerType)

	// Create a container with the correct ports and start command
	webport := 8081
	webp := ocon.ContainerPort(webPortName, webport)
	cont := ocon.Container(dc.Name, image).
		Ports(webp).
		SetLivenessProbe(probes.NewHTTPGetProbe(webport)).EnvVars(makeWorkerEnvVars(clustername, sparkconfdir))

	if sparkworkerconfig != "" {
		pt = pt.SetConfigMapVolume(sparkworkerconfig)
		cont = cont.SetVolumeMount(sparkworkerconfig, sparkconfdir, true)
	}

	// Finally, assign the container to the pod template spec and
	// assign the pod template spec to the deployment config
	return dc.PodTemplateSpec(pt.Containers(cont))
}

func mastername(clustername string) string {
	return clustername + "-m"
}

func workername(clustername string) string {
	return clustername + "-w"
}

func sparkMaster(namespace, image string, replicas int, clustername, sparkconfdir, sparkmasterconfig, driverdc string) *odc.ODeploymentConfig {

	// Create the basic deployment config
	// We will use a label and pod selector based on the cluster name
	// Openshift will add additional labels and selectors to distinguish pods handled by
	// this deploymentconfig from pods beloning to another.
	dc := odc.DeploymentConfig(mastername(clustername), namespace).
		TriggerOnConfigChange().RollingStrategy().Label(clusterLabel, clustername).
		Label(typeLabel, masterType).
		PodSelector(clusterLabel, clustername).Replicas(replicas)

	if driverdc != "" {
		dc = dc.Label(ephemeralLabel, driverdc)
	}

	// Create a pod template spec with the matching label
	pt := opt.PodTemplateSpec().Label(clusterLabel, clustername).
		Label(typeLabel, masterType)

	// Create a container with the correct ports and start command
	liveness := probes.NewExecProbe([]string{"/bin/bash", "-c", "curl localhost:8080 | grep -e Status.*ALIVE"})
	liveness.InitialDelaySeconds = 10
	readiness := probes.NewHTTPGetProbe(webPort)
	masterp := ocon.ContainerPort(masterPortName, masterPort)
	webp := ocon.ContainerPort(webPortName, webPort)
	cont := ocon.Container(dc.Name, image).
		Ports(masterp, webp).
		SetLivenessProbe(liveness).
		SetReadinessProbe(readiness).EnvVars(makeEnvVars(clustername, sparkconfdir))

	if sparkmasterconfig != "" {
		pt = pt.SetConfigMapVolume(sparkmasterconfig)
		cont = cont.SetVolumeMount(sparkmasterconfig, sparkconfdir, true)
	}

	// Finally, assign the container to the pod template spec and
	// assign the pod template spec to the deployment config
	return dc.PodTemplateSpec(pt.Containers(cont))
}

func service(name string,
	port int,
	clustername, otype string,
	podselectors map[string]string) (*osv.OService, *osv.OServicePort) {

	p := osv.ServicePort(port).TargetPort(port)
	return osv.Service(name).Label(clusterLabel, clustername).
		Label(typeLabel, otype).PodSelectors(podselectors).Ports(p), p
}

func checkForConfigMap(name string, cm kclient.ConfigMapsInterface) error {
	_, err := cm.Get(name)
	if err != nil {
		if strings.Index(err.Error(), "not found") != -1 {
			return generalErr(err, fmt.Sprintf(missingConfigMsg, name), ClusterConfigCode)
		}
		return generalErr(nil, fmt.Sprintf(missingConfigMsg, name), ClientOperationCode)
	}
	return nil
}

func countWorkers(client kclient.PodInterface, clustername string) (int, *kapi.PodList, error) {
	// If we are  unable to retrieve a list of worker pods, return -1 for count
	// This is an error case, differnt from a list of length 0. Let the caller
	// decide whether to report the error or the -1 count
	cnt := -1
	selectorlist := makeSelector(workerType, clustername)
	pods, err := client.List(selectorlist)
	if pods != nil {
		cnt = len(pods.Items)
	}
	return cnt, pods, err
}

func getDriverDeployment(app, namespace string, client *kclient.Client) string {

	// When we make calls from a driver pod, the most likely value we have is a deployment
	// so use that first
	rcc := client.ReplicationControllers(namespace)
	rc, err := rcc.Get(app)
	if err == nil && rc != nil {
		return app
	}

	// Okay, it wasn't a deployment, maybe it's a pod
	pc := client.Pods(namespace)
	pod, err := pc.Get(app)
	if err == nil && pod != nil {
		return pod.Labels["deployment"]
	}
	return ""
}

// Create a cluster and return the representation
func CreateCluster(
	clustername, namespace, sparkimage string,
	config *ClusterConfig, osclient *oclient.Client, client kclient.Interface, app string, ephemeral bool) (SparkCluster, error) {

	var driverrc string
	var ephem_val string
	var masterconfdir string
	var workerconfdir string
	var result SparkCluster = SparkCluster{}

	createCode := func(err error) int {
		if err != nil && strings.Index(err.Error(), "already exists") != -1 {
			return ComponentExistsCode
		}
		return ClientOperationCode
	}

	masterhost := clustername

	// Check to see if a cluster already exists of the same name (complete or incomplete)
	existing := SparkCluster{}
	findClusterBody(clustername, namespace, osclient, client, &existing)
	if !CheckNoCluster(&existing) {
		var msg string
		if existing.Status != "Running" {
			msg = fmt.Sprintf(clusterExistsMsg, clustername, " (incomplete)")
		} else {
			msg = fmt.Sprintf(clusterExistsMsg, clustername, "")
		}
		return result, generalErr(nil, msg, ComponentExistsCode)
	}


	// Copy any named config referenced and update it with any explicit config values
	finalconfig, err := GetClusterConfig(config, client.ConfigMaps(namespace))
	if err != nil {
		return result, generalErr(err, clusterConfigMsg, ErrorCode(err))
	}
	if finalconfig.SparkImage != "" {
		sparkimage = finalconfig.SparkImage
	} else if sparkimage == "" {
		return result, generalErr(nil, sparkImageMsg, ClusterConfigCode)
	}

	mastercount := int(finalconfig.MasterCount)
	workercount := int(finalconfig.WorkerCount)

	// Check if finalconfig contains the names of ConfigMaps to use for spark
	// configuration. If so they must exist. The ConfigMaps will be mounted
	// as volumes on spark pods and the path stored in the environment
	// variable UPDATE_SPARK_CONF_DIR
	cm := client.ConfigMaps(namespace)
	if finalconfig.SparkMasterConfig != "" {
		err := checkForConfigMap(finalconfig.SparkMasterConfig, cm)
		if err != nil {
			return result, err
		}
		masterconfdir = sparkconfdir
	}

	if finalconfig.SparkWorkerConfig != "" {
		err := checkForConfigMap(finalconfig.SparkWorkerConfig, cm)
		if err != nil {
			return result, err
		}
		workerconfdir = sparkconfdir
	}

	// If an app value was passed find the deployment so we can do the cluster
	// association and potentially mark the cluster as ephemeral
	if app != "" {
		driverrc = getDriverDeployment(app, namespace, client)
	}
	if ephemeral {
		// If we can't find an rc then we just make a long-running cluster
		ephem_val = driverrc
	}

	// Create the master deployment config
	masterdc := sparkMaster(namespace, sparkimage, mastercount, clustername, masterconfdir, finalconfig.SparkMasterConfig, ephem_val)

	// Create the services that will be associated with the master pod
	// They will be created with selectors based on the pod labels
	mastersv, _ := service(masterhost,
		masterdc.FindPort(masterPortName),
		clustername, masterType,
		masterdc.GetPodTemplateSpecLabels())

	websv, _ := service(masterhost+webServiceSuffix,
		masterdc.FindPort(webPortName),
		clustername, webuiType,
		masterdc.GetPodTemplateSpecLabels())

	webuiroute := ort.NewRoute(websv.GetName() + "-route", websv.GetName(), clustername, "webui")

	// Create the worker deployment config
	workerdc := sparkWorker(namespace, sparkimage, workercount, clustername, workerconfdir, finalconfig.SparkWorkerConfig)

	// Launch all of the objects
	dcc := osclient.DeploymentConfigs(namespace)
	_, err = dcc.Create(&masterdc.DeploymentConfig)
	if err != nil {
		return result, generalErr(err, fmt.Sprintf(createDepConfigMsg, masterdc.Name), createCode(err))
	}
	_, err = dcc.Create(&workerdc.DeploymentConfig)
	if err != nil {
		// Since we created the master deployment config, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, fmt.Sprintf(createDepConfigMsg, workerdc.Name), createCode(err))
	}

	sc := client.Services(namespace)
	_, err = sc.Create(&mastersv.Service)
	if err != nil {
		// Since we've created things, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, masterSrvMsg, createCode(err))
	}


	_, err = sc.Create(&websv.Service)
	if err != nil {
		// Since we've created things, try to clean up
		DeleteCluster(clustername, namespace, osclient, client, "", "")
		return result, generalErr(err, masterWebSrvMsg, createCode(err))
	}

	// We will expose the Spark master webui unless we are told not to do it
	if config.ExposeWebUI {
		rc := osclient.Routes(namespace)
		_, err = rc.Create(webuiroute)
	}

	// Wait for the replication controllers to exist before building the response.
	rcc := client.ReplicationControllers(namespace)
	{
		var mrepl, wrepl *kapi.ReplicationController
		mrepl = nil
		wrepl = nil
		for i := 0; i < 4; i++ {
			if mrepl == nil {
				mrepl, _ = getReplController(rcc, clustername, masterType)
			}
			if wrepl == nil {
				wrepl, _ = getReplController(rcc, clustername, workerType)
			}
			if wrepl != nil && mrepl != nil {
				break
			}
			time.Sleep(250 * time.Millisecond)
		}
	}

	// Now that the creation actually worked, label the dc if the app value was passed.
	// Note that updates can fail if someone updates the object underneath us, so
	// we have to try again.  Try for 5 seconds
	if driverrc != "" {
		for i := 0; i < 20; i++ {
			driver, err := rcc.Get(driverrc)
			if err == nil {
				if driver.Labels == nil {
					driver.Labels = map[string]string{}
				}
				driver.Labels[driverLabel] = clustername
				_, err = rcc.Update(driver)
				if err == nil {
					break
				}
				time.Sleep(250 * time.Millisecond)
			} else {
				break
			}
		}
	}

{'fraction_non_alphanumeric': 0.11148403483309144, 'fraction_numerical': 0.002751572327044025, 'mean_word_length': 4.600846740050804, 'pattern_counts': {'":': 6, '<': 8, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '4520845', 'n_tokens_mistral': 11086, 'n_tokens_neox': 10479, 'n_words': 2959}