<commit_before>package server

import (
	"context"
	"encoding/json"
	"strconv"

	client "github.com/pachyderm/pachyderm/src/client"
	"github.com/pachyderm/pachyderm/src/client/enterprise"
	"github.com/pachyderm/pachyderm/src/client/pps"
	"github.com/pachyderm/pachyderm/src/client/version"
	"github.com/pachyderm/pachyderm/src/server/pkg/deploy/assets"
	"github.com/pachyderm/pachyderm/src/server/pkg/ppsutil"
	"github.com/pachyderm/pachyderm/src/server/worker"

	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
)

// Parameters used when creating the kubernetes replication controller in charge
// of a job or pipeline's workers
type workerOptions struct {
	rcName string // Name of the replication controller managing workers

	userImage        string              // The user's pipeline/job image
	labels           map[string]string   // k8s labels attached to the RC and workers
	annotations      map[string]string   // k8s annotations attached to the RC and workers
	parallelism      int32               // Number of replicas the RC maintains
	cacheSize        string              // Size of cache that sidecar uses
	resourceRequests *v1.ResourceList    // Resources requested by pipeline/job pods
	resourceLimits   *v1.ResourceList    // Resources requested by pipeline/job pods
	workerEnv        []v1.EnvVar         // Environment vars set in the user container
	volumes          []v1.Volume         // Volumes that we expose to the user container
	volumeMounts     []v1.VolumeMount    // Paths where we mount each volume in 'volumes'
	etcdPrefix       string              // the prefix in etcd to use
	schedulingSpec   *pps.SchedulingSpec // the SchedulingSpec for the pipeline
	podSpec          string

	// Secrets that we mount in the worker container (e.g. for reading/writing to
	// s3)
	imagePullSecrets []v1.LocalObjectReference
	service          *pps.Service
}

func (a *apiServer) workerPodSpec(options *workerOptions) (v1.PodSpec, error) {
	pullPolicy := a.workerImagePullPolicy
	if pullPolicy == "" {
		pullPolicy = "IfNotPresent"
	}
	sidecarEnv := []v1.EnvVar{{
		Name:  "BLOCK_CACHE_BYTES",
		Value: options.cacheSize,
	}, {
		Name:  "PFS_CACHE_SIZE",
		Value: "16",
	}, {
		Name:  "PACH_ROOT",
		Value: a.storageRoot,
	}, {
		Name:  "STORAGE_BACKEND",
		Value: a.storageBackend,
	}}
	// This only happens in local deployment.  We want the workers to be
	// able to read from/write to the hostpath volume as well.
	storageVolumeName := "pach-disk"
	var sidecarVolumeMounts []v1.VolumeMount
	if a.storageHostPath != "" {
		options.volumes = append(options.volumes, v1.Volume{
			Name: storageVolumeName,
			VolumeSource: v1.VolumeSource{
				HostPath: &v1.HostPathVolumeSource{
					Path: a.storageHostPath,
				},
			},
		})

		sidecarVolumeMounts = []v1.VolumeMount{
			{
				Name:      storageVolumeName,
				MountPath: a.storageRoot,
			},
		}
	}
	userVolumeMounts := options.volumeMounts
	secretVolume, secretMount := assets.GetBackendSecretVolumeAndMount(a.storageBackend)
	options.volumes = append(options.volumes, secretVolume)
	options.volumeMounts = append(options.volumeMounts, secretMount)
	sidecarVolumeMounts = append(sidecarVolumeMounts, secretMount)
	userVolumeMounts = append(userVolumeMounts, secretMount)

	// Explicitly set CPU, MEM and DISK requests to zero because some cloud
	// providers set their own defaults which are usually not what we want.
	cpuZeroQuantity := resource.MustParse("0")
	memZeroQuantity := resource.MustParse("0M")
	memSidecarQuantity := resource.MustParse(options.cacheSize)

	if !a.noExposeDockerSocket {
		options.volumes = append(options.volumes, v1.Volume{
			Name: "docker",
			VolumeSource: v1.VolumeSource{
				HostPath: &v1.HostPathVolumeSource{
					Path: "/var/run/docker.sock",
				},
			},
		})
		userVolumeMounts = append(userVolumeMounts, v1.VolumeMount{
			Name:      "docker",
			MountPath: "/var/run/docker.sock",
		})
	}
	zeroVal := int64(0)
	workerImage := a.workerImage
	resp, err := a.getPachClient().Enterprise.GetState(context.Background(), &enterprise.GetStateRequest{})
	if err != nil {
		return v1.PodSpec{}, err
	}
	if resp.State != enterprise.State_ACTIVE {
		workerImage = assets.AddRegistry("", workerImage)
	}
	podSpec := v1.PodSpec{
		InitContainers: []v1.Container{
			{
				Name:            "init",
				Image:           workerImage,
				Command:         []string{"/pach/worker.sh"},
				ImagePullPolicy: v1.PullPolicy(pullPolicy),
				Env:             options.workerEnv,
				VolumeMounts:    options.volumeMounts,
			},
		},
		Containers: []v1.Container{
			{
				Name:            client.PPSWorkerUserContainerName,
				Image:           options.userImage,
				Command:         []string{"/pach-bin/worker"},
				ImagePullPolicy: v1.PullPolicy(pullPolicy),
				Env:             options.workerEnv,
				VolumeMounts:    userVolumeMounts,<commit_msg>Remove env vars from init container.<commit_after>package server

import (
	"context"
	"encoding/json"
	"strconv"

	client "github.com/pachyderm/pachyderm/src/client"
	"github.com/pachyderm/pachyderm/src/client/enterprise"
	"github.com/pachyderm/pachyderm/src/client/pps"
	"github.com/pachyderm/pachyderm/src/client/version"
	"github.com/pachyderm/pachyderm/src/server/pkg/deploy/assets"
	"github.com/pachyderm/pachyderm/src/server/pkg/ppsutil"
	"github.com/pachyderm/pachyderm/src/server/worker"

	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
)

// Parameters used when creating the kubernetes replication controller in charge
// of a job or pipeline's workers
type workerOptions struct {
	rcName string // Name of the replication controller managing workers

	userImage        string              // The user's pipeline/job image
	labels           map[string]string   // k8s labels attached to the RC and workers
	annotations      map[string]string   // k8s annotations attached to the RC and workers
	parallelism      int32               // Number of replicas the RC maintains
	cacheSize        string              // Size of cache that sidecar uses
	resourceRequests *v1.ResourceList    // Resources requested by pipeline/job pods
	resourceLimits   *v1.ResourceList    // Resources requested by pipeline/job pods
	workerEnv        []v1.EnvVar         // Environment vars set in the user container
	volumes          []v1.Volume         // Volumes that we expose to the user container
	volumeMounts     []v1.VolumeMount    // Paths where we mount each volume in 'volumes'
	etcdPrefix       string              // the prefix in etcd to use
	schedulingSpec   *pps.SchedulingSpec // the SchedulingSpec for the pipeline
	podSpec          string

	// Secrets that we mount in the worker container (e.g. for reading/writing to
	// s3)
	imagePullSecrets []v1.LocalObjectReference
	service          *pps.Service
}

func (a *apiServer) workerPodSpec(options *workerOptions) (v1.PodSpec, error) {
	pullPolicy := a.workerImagePullPolicy
	if pullPolicy == "" {
		pullPolicy = "IfNotPresent"
	}
	sidecarEnv := []v1.EnvVar{{
		Name:  "BLOCK_CACHE_BYTES",
		Value: options.cacheSize,
	}, {
		Name:  "PFS_CACHE_SIZE",
		Value: "16",
	}, {
		Name:  "PACH_ROOT",
		Value: a.storageRoot,
	}, {
		Name:  "STORAGE_BACKEND",
		Value: a.storageBackend,
	}}
	// This only happens in local deployment.  We want the workers to be
	// able to read from/write to the hostpath volume as well.
	storageVolumeName := "pach-disk"
	var sidecarVolumeMounts []v1.VolumeMount
	if a.storageHostPath != "" {
		options.volumes = append(options.volumes, v1.Volume{
			Name: storageVolumeName,
			VolumeSource: v1.VolumeSource{
				HostPath: &v1.HostPathVolumeSource{
					Path: a.storageHostPath,
				},
			},
		})

		sidecarVolumeMounts = []v1.VolumeMount{
			{
				Name:      storageVolumeName,
				MountPath: a.storageRoot,
			},
		}
	}
	userVolumeMounts := options.volumeMounts
	secretVolume, secretMount := assets.GetBackendSecretVolumeAndMount(a.storageBackend)
	options.volumes = append(options.volumes, secretVolume)
	options.volumeMounts = append(options.volumeMounts, secretMount)
	sidecarVolumeMounts = append(sidecarVolumeMounts, secretMount)
	userVolumeMounts = append(userVolumeMounts, secretMount)

	// Explicitly set CPU, MEM and DISK requests to zero because some cloud
	// providers set their own defaults which are usually not what we want.
	cpuZeroQuantity := resource.MustParse("0")
	memZeroQuantity := resource.MustParse("0M")
	memSidecarQuantity := resource.MustParse(options.cacheSize)

	if !a.noExposeDockerSocket {
		options.volumes = append(options.volumes, v1.Volume{
			Name: "docker",
			VolumeSource: v1.VolumeSource{
				HostPath: &v1.HostPathVolumeSource{
					Path: "/var/run/docker.sock",
				},
			},
		})
		userVolumeMounts = append(userVolumeMounts, v1.VolumeMount{
			Name:      "docker",
			MountPath: "/var/run/docker.sock",
		})
	}
	zeroVal := int64(0)
	workerImage := a.workerImage
	resp, err := a.getPachClient().Enterprise.GetState(context.Background(), &enterprise.GetStateRequest{})
	if err != nil {
		return v1.PodSpec{}, err
	}
	if resp.State != enterprise.State_ACTIVE {
		workerImage = assets.AddRegistry("", workerImage)
	}
	podSpec := v1.PodSpec{
		InitContainers: []v1.Container{
			{
				Name:            "init",
				Image:           workerImage,
				Command:         []string{"/pach/worker.sh"},
				ImagePullPolicy: v1.PullPolicy(pullPolicy),
				VolumeMounts:    options.volumeMounts,
			},
		},
		Containers: []v1.Container{
			{
				Name:            client.PPSWorkerUserContainerName,
				Image:           options.userImage,
				Command:         []string{"/pach-bin/worker"},
				ImagePullPolicy: v1.PullPolicy(pullPolicy),
				Env:             options.workerEnv,
				VolumeMounts:    userVolumeMounts,
{'fraction_non_alphanumeric': 0.10774683544303798, 'fraction_numerical': 0.008708860759493672, 'mean_word_length': 3.3239929947460594, 'pattern_counts': {'":': 0, '<': 3, '<?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': '1766898', 'n_tokens_mistral': 3501, 'n_tokens_neox': 3183, 'n_words': 724}