<commit_before>package pipeline

import (
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"

	"encoding/json"

	"github.com/AcalephStorage/kontinuous/kube"
	"github.com/Sirupsen/logrus"
)

// CreateJob creates a kubernetes Job for the given build information
func CreateJob(definition *Definition, jobInfo *JobBuildInfo) (j *kube.Job, err error) {

	newJob, _ := build(definition, jobInfo)

	err = deployJob(newJob)
	if err != nil {
		logrus.WithError(err).Errorln("Unable to Create Job")
		return nil, err
	}

	return newJob, nil
}

func GetJobBuildInfo(jobInfo []byte) (payload *JobBuildInfo, err error) {

	if len(jobInfo) == 0 {
		return nil, errors.New("Empty JSON String")
	}

	if err = json.Unmarshal(jobInfo, &payload); err != nil {
		return nil, err
	}

	return payload, nil
}

func build(definition *Definition, jobInfo *JobBuildInfo) (j *kube.Job, err error) {

	namespace := getNamespace(definition)
	name := fmt.Sprintf("%s-%s-%s", jobInfo.PipelineUUID, jobInfo.Build, jobInfo.Stage)
	j = kube.NewJob(name, namespace)

	addJobDetail(j, definition, jobInfo)
	addSpecDetails(j, definition, jobInfo)
	return j, nil

}

func addJobDetail(j *kube.Job, definition *Definition, jobInfo *JobBuildInfo) {

	selectors := map[string]string{
		"pipeline": jobInfo.PipelineUUID,
		"build":    jobInfo.Build,
		"stage":    jobInfo.Stage,
	}

	for key, value := range selectors {
		j.AddSelectorMatchLabel(key, value)
	}
}

func addSpecDetails(j *kube.Job, definitions *Definition, jobInfo *JobBuildInfo) {

	stage := getCurrentStage(definitions, jobInfo)

	source := j.AddPodVolume("kontinuous-source", "/kontinuous/src")
	status := j.AddPodVolume("kontinuous-status", "/kontinuous/status")
	docker := j.AddPodVolume("kontinuous-docker", "/var/run/docker.sock")
	secrets := getSecrets(definitions.Spec.Template.Secrets, getNamespace(definitions))

	agentContainer := createAgentContainer(definitions, jobInfo)
	agentContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
	agentContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
	agentContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
	setContainerEnv(agentContainer, secrets)
	addJobContainer(j, agentContainer)

	switch stage.Type {
	case "docker_build":

		dockerContainer := createDockerContainer(stage, jobInfo, "BUILD")
		dockerContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		dockerContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		dockerContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(dockerContainer, secrets)
		addJobContainer(j, dockerContainer)

	case "docker_publish":
		dockerContainer := createDockerContainer(stage, jobInfo, "PUBLISH")
		dockerContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		dockerContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		dockerContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(dockerContainer, secrets)
		addJobContainer(j, dockerContainer)

	case "command":
		commandContainer := createCommandContainer(stage, jobInfo)
		commandContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		commandContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		commandContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(commandContainer, secrets)
		addJobContainer(j, commandContainer)
	}

}

func getCurrentStage(definitions *Definition, jobInfo *JobBuildInfo) (stage *Stage) {

	index, _ := strconv.Atoi(jobInfo.Stage)

	if currentIndex := index - 1; 0 <= currentIndex && currentIndex < len(definitions.Spec.Template.Stages) {
		return &definitions.Spec.Template.Stages[currentIndex]
	}

	return &Stage{}
}

func createAgentContainer(definitions *Definition, jobInfo *JobBuildInfo) *kube.Container {

	container := createJobContainer("kontinuous-agent", "quay.io/acaleph/kontinuous-agent:latest")
	envVars := map[string]string{
		"REQUIRE_SOURCE_CODE": "TRUE",
		"GIT_COMMIT":          jobInfo.Commit,
		"GIT_USER":            jobInfo.User,
		"GIT_REPO":            jobInfo.Repo,
		"GIT_OWNER":           jobInfo.Owner,
		"PIPELINE_ID":         jobInfo.PipelineUUID,
		"BUILD_ID":            jobInfo.Build,
		"STAGE_ID":            jobInfo.Stage,
		"S3_URL":              os.Getenv("S3_URL"),
		"S3_ACCESS_KEY":       os.Getenv("S3_ACCESS_KEY"),
		"S3_SECRET_KEY":       os.Getenv("S3_SECRET_KEY"),
		"KONTINUOUS_URL":      os.Getenv("KONTINUOUS_URL"),
		"NAMESPACE":           getNamespace(definitions),
		"ARTIFACT_URL":        "",
	}

	setContainerEnv(container, envVars)
	return container
}

func createDockerContainer(stage *Stage, jobInfo *JobBuildInfo, mode string) *kube.Container {
	imageName := fmt.Sprintf("%s-%s", jobInfo.PipelineUUID, jobInfo.Build)
	container := createJobContainer("docker-agent", "quay.io/acaleph/docker-agent:latest")

	envVar := map[string]string{
		"INTERNAL_REGISTRY":   os.Getenv("INTERNAL_REGISTRY"),
		"DOCKERFILE_NAME":     "Dockerfile",
		"DOCKERFILE_PATH":     ".",
		"REQUIRE_CREDENTIALS": "TRUE",
		"IMAGE_NAME":          imageName,
		"MODE":                mode,
		"PIPELINE_ID":         jobInfo.PipelineUUID,
		"BUILD_ID":            jobInfo.Build,
		"STAGE_ID":            jobInfo.Stage,
		"IMAGE_TAG":           jobInfo.Commit,
		"BRANCH":              jobInfo.Branch,
	}

	for stageEnvKey, stageEnvValue := range stage.Params {
		envVar[strings.ToUpper(stageEnvKey)] = stageEnvValue.(string)
	}

	setContainerEnv(container, envVar)
	return container
}

func createCommandContainer(stage *Stage, jobInfo *JobBuildInfo) *kube.Container {

	containerName := fmt.Sprintf("%s-%s", jobInfo.PipelineUUID, jobInfo.Build)
	cmdImage := fmt.Sprintf("%s/%s:%s", os.Getenv("INTERNAL_REGISTRY"), containerName, jobInfo.Commit)
	imageName := "quay.io/acaleph/command-agent:latest"
	container := createJobContainer(containerName, imageName)
	container.Image = imageName
	container.AddEnv("IMAGE", cmdImage)
	container.WorkingDir = fmt.Sprintf("/kontinuous/src/%v/%v/%v", jobInfo.PipelineUUID, jobInfo.Build, stage.Index)

	for paramKey, paramValue := range stage.Params {

		switch strings.ToUpper(paramKey) {
		case "COMMAND":
			commands := paramValue.([]interface{})
			stringCommand := make([]string, len(commands))
			for i, c := range commands {
				stringCommand[i] = c.(string)
			}
			container.AddEnv("COMMAND", strings.Join(stringCommand, " "))
		case "ARGS":
			args := paramValue.([]interface{})
			stringArg := make([]string, len(args))
			for i, a := range args {
				stringArg[i] = a.(string)
			}
			container.SetArgs(stringArg...)
		case "IMAGE":
			container.AddEnv("IMAGE", paramValue.(string))
		case "WORKING_DIR":
			container.WorkingDir = paramValue.(string)
			container.AddEnv("WORKING_DIR", paramValue.(string))<commit_msg>updated command-agent container name<commit_after>package pipeline

import (
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"

	"encoding/json"

	"github.com/AcalephStorage/kontinuous/kube"
	"github.com/Sirupsen/logrus"
)

// CreateJob creates a kubernetes Job for the given build information
func CreateJob(definition *Definition, jobInfo *JobBuildInfo) (j *kube.Job, err error) {

	newJob, _ := build(definition, jobInfo)

	err = deployJob(newJob)
	if err != nil {
		logrus.WithError(err).Errorln("Unable to Create Job")
		return nil, err
	}

	return newJob, nil
}

func GetJobBuildInfo(jobInfo []byte) (payload *JobBuildInfo, err error) {

	if len(jobInfo) == 0 {
		return nil, errors.New("Empty JSON String")
	}

	if err = json.Unmarshal(jobInfo, &payload); err != nil {
		return nil, err
	}

	return payload, nil
}

func build(definition *Definition, jobInfo *JobBuildInfo) (j *kube.Job, err error) {

	namespace := getNamespace(definition)
	name := fmt.Sprintf("%s-%s-%s", jobInfo.PipelineUUID, jobInfo.Build, jobInfo.Stage)
	j = kube.NewJob(name, namespace)

	addJobDetail(j, definition, jobInfo)
	addSpecDetails(j, definition, jobInfo)
	return j, nil

}

func addJobDetail(j *kube.Job, definition *Definition, jobInfo *JobBuildInfo) {

	selectors := map[string]string{
		"pipeline": jobInfo.PipelineUUID,
		"build":    jobInfo.Build,
		"stage":    jobInfo.Stage,
	}

	for key, value := range selectors {
		j.AddSelectorMatchLabel(key, value)
	}
}

func addSpecDetails(j *kube.Job, definitions *Definition, jobInfo *JobBuildInfo) {

	stage := getCurrentStage(definitions, jobInfo)

	source := j.AddPodVolume("kontinuous-source", "/kontinuous/src")
	status := j.AddPodVolume("kontinuous-status", "/kontinuous/status")
	docker := j.AddPodVolume("kontinuous-docker", "/var/run/docker.sock")
	secrets := getSecrets(definitions.Spec.Template.Secrets, getNamespace(definitions))

	agentContainer := createAgentContainer(definitions, jobInfo)
	agentContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
	agentContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
	agentContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
	setContainerEnv(agentContainer, secrets)
	addJobContainer(j, agentContainer)

	switch stage.Type {
	case "docker_build":

		dockerContainer := createDockerContainer(stage, jobInfo, "BUILD")
		dockerContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		dockerContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		dockerContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(dockerContainer, secrets)
		addJobContainer(j, dockerContainer)

	case "docker_publish":
		dockerContainer := createDockerContainer(stage, jobInfo, "PUBLISH")
		dockerContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		dockerContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		dockerContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(dockerContainer, secrets)
		addJobContainer(j, dockerContainer)

	case "command":
		commandContainer := createCommandContainer(stage, jobInfo)
		commandContainer.AddVolumeMountPoint(source, "/kontinuous/src", false)
		commandContainer.AddVolumeMountPoint(status, "/kontinuous/status", false)
		commandContainer.AddVolumeMountPoint(docker, "/var/run/docker.sock", false)
		setContainerEnv(commandContainer, secrets)
		addJobContainer(j, commandContainer)
	}

}

func getCurrentStage(definitions *Definition, jobInfo *JobBuildInfo) (stage *Stage) {

	index, _ := strconv.Atoi(jobInfo.Stage)

	if currentIndex := index - 1; 0 <= currentIndex && currentIndex < len(definitions.Spec.Template.Stages) {
		return &definitions.Spec.Template.Stages[currentIndex]
	}

	return &Stage{}
}

func createAgentContainer(definitions *Definition, jobInfo *JobBuildInfo) *kube.Container {

	container := createJobContainer("kontinuous-agent", "quay.io/acaleph/kontinuous-agent:latest")
	envVars := map[string]string{
		"REQUIRE_SOURCE_CODE": "TRUE",
		"GIT_COMMIT":          jobInfo.Commit,
		"GIT_USER":            jobInfo.User,
		"GIT_REPO":            jobInfo.Repo,
		"GIT_OWNER":           jobInfo.Owner,
		"PIPELINE_ID":         jobInfo.PipelineUUID,
		"BUILD_ID":            jobInfo.Build,
		"STAGE_ID":            jobInfo.Stage,
		"S3_URL":              os.Getenv("S3_URL"),
		"S3_ACCESS_KEY":       os.Getenv("S3_ACCESS_KEY"),
		"S3_SECRET_KEY":       os.Getenv("S3_SECRET_KEY"),
		"KONTINUOUS_URL":      os.Getenv("KONTINUOUS_URL"),
		"NAMESPACE":           getNamespace(definitions),
		"ARTIFACT_URL":        "",
	}

	setContainerEnv(container, envVars)
	return container
}

func createDockerContainer(stage *Stage, jobInfo *JobBuildInfo, mode string) *kube.Container {
	imageName := fmt.Sprintf("%s-%s", jobInfo.PipelineUUID, jobInfo.Build)
	container := createJobContainer("docker-agent", "quay.io/acaleph/docker-agent:latest")

	envVar := map[string]string{
		"INTERNAL_REGISTRY":   os.Getenv("INTERNAL_REGISTRY"),
		"DOCKERFILE_NAME":     "Dockerfile",
		"DOCKERFILE_PATH":     ".",
		"REQUIRE_CREDENTIALS": "TRUE",
		"IMAGE_NAME":          imageName,
		"MODE":                mode,
		"PIPELINE_ID":         jobInfo.PipelineUUID,
		"BUILD_ID":            jobInfo.Build,
		"STAGE_ID":            jobInfo.Stage,
		"IMAGE_TAG":           jobInfo.Commit,
		"BRANCH":              jobInfo.Branch,
	}

	for stageEnvKey, stageEnvValue := range stage.Params {
		envVar[strings.ToUpper(stageEnvKey)] = stageEnvValue.(string)
	}

	setContainerEnv(container, envVar)
	return container
}

func createCommandContainer(stage *Stage, jobInfo *JobBuildInfo) *kube.Container {

	containerName := "command-agent"
	cmdImage := fmt.Sprintf("%s/%s:%s", os.Getenv("INTERNAL_REGISTRY"), containerName, jobInfo.Commit)
	imageName := "quay.io/acaleph/command-agent:latest"
	container := createJobContainer(containerName, imageName)
	container.Image = imageName
	container.AddEnv("IMAGE", cmdImage)
	container.WorkingDir = fmt.Sprintf("/kontinuous/src/%s/%s/%d", jobInfo.PipelineUUID, jobInfo.Build, stage.Index)

	for paramKey, paramValue := range stage.Params {

		switch strings.ToUpper(paramKey) {
		case "COMMAND":
			commands := paramValue.([]interface{})
			stringCommand := make([]string, len(commands))
			for i, c := range commands {
				stringCommand[i] = c.(string)
			}
			container.AddEnv("COMMAND", strings.Join(stringCommand, " "))
		case "ARGS":
			args := paramValue.([]interface{})
			stringArg := make([]string, len(args))
			for i, a := range args {
				stringArg[i] = a.(string)
			}
			container.SetArgs(stringArg...)
		case "IMAGE":
			container.AddEnv("IMAGE", paramValue.(string))
		case "WORKING_DIR":
			container.WorkingDir = paramValue.(string)
			container.AddEnv("WORKING_DIR", paramValue.(string))
{'fraction_non_alphanumeric': 0.13574527538789616, 'fraction_numerical': 0.0013236267372600927, 'mean_word_length': 4.802047781569966, 'pattern_counts': {'":': 70, '<': 7, '<?xml version=': 0, '>': 3, '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': '3477487', 'n_tokens_mistral': 4768, 'n_tokens_neox': 4484, 'n_words': 820}