<commit_before>// Copyright 2018 The Kubeflow 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 trainer is to manage TensorFlow training jobs.
package trainer

import (
	"fmt"
	"reflect"
	"strings"

	log "github.com/sirupsen/logrus"
	"k8s.io/api/core/v1"
	"k8s.io/api/policy/v1beta1"
	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/record"

	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/helper"
	tfv1alpha1 "github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1alpha1"
	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/validation"
	tfjobclient "github.com/kubeflow/tf-operator/pkg/client/clientset/versioned"
	"github.com/kubeflow/tf-operator/pkg/client/clientset/versioned/scheme"
	"github.com/kubeflow/tf-operator/pkg/util"
	train_util "github.com/kubeflow/tf-operator/pkg/util/train"
)

// TODO(jlewi): We should switch a New pattern and make trainingJob private so we can
// TrainingJob represents a training job specification.
// ensure correctness on creation.
type TrainingJob struct {
	job *tfv1alpha1.TFJob

	KubeCli kubernetes.Interface

	recorder record.EventRecorder

	Replicas []*TFReplicaSet

	tfJobClient tfjobclient.Interface

	// in memory state of the job.
	// status is the source of truth after job struct is materialized. Changes to the status to be persisted
	// should be made here.
	status tfv1alpha1.TFJobStatus

	memberCounter int

	pdb *v1beta1.PodDisruptionBudget

	// contextLogger is a logger to use for logging information about this replica.
	contextLogger *log.Entry
}

// ClusterSpec represents a cluster TensorFlow specification.
// https://www.tensorflow.org/deploy/distributed#create_a_tftrainclusterspec_to_describe_the_cluster
// It is a map from job names to network addresses.
type ClusterSpec map[string][]string

// TaskSpec represents a Task specification.
type TaskSpec struct {
	Type  string `json:"type"`
	Index int    `json:"index"`
}

//initJob initiate a training job and returns the job specifications.
func initJob(kubeCli kubernetes.Interface, tfJobClient tfjobclient.Interface, recorder record.EventRecorder, job *tfv1alpha1.TFJob) (*TrainingJob, error) {
	j := &TrainingJob{
		KubeCli:     kubeCli,
		tfJobClient: tfJobClient,
		recorder:    recorder,
		Replicas:    make([]*TFReplicaSet, 0),
		job:         job,
		status:      *job.Status.DeepCopy(),

		contextLogger: log.WithFields(log.Fields{
			// We use job to match the key used in controller.go
			// In controller.go we log the key used with the workqueue.
			"job": job.ObjectMeta.Namespace + "/" + job.ObjectMeta.Name,
			"uid": job.ObjectMeta.UID,
		}),
	}

	return j, nil
}

// NewJob method invokes the initJob and check for error
func NewJob(kubeCli kubernetes.Interface, tfJobClient tfjobclient.Interface, recorder record.EventRecorder, job *tfv1alpha1.TFJob, config *tfv1alpha1.ControllerConfig) (*TrainingJob, error) {
	j, err := initJob(kubeCli, tfJobClient, recorder, job)
	if err != nil {
		return nil, err
	}

	return j, nil
}

// Update replaces the TFJob corresponding to TrainingJob with the provided job.
// This function is used when the Spec/Status of the job is modified outside the controller.
// For example, if the user issues a delete request. This will update the metadata on the object
// so we need to replace the spec.
func (j *TrainingJob) Update(newJob *tfv1alpha1.TFJob) {
	j.contextLogger.Info("Updating job to %+v", *newJob)
	j.job = newJob
}

// UID returns the user ID of the requesting user
func (j *TrainingJob) UID() types.UID {
	return j.job.ObjectMeta.UID
}

// ClusterSpec returns the cluster specification for the training job provided
func (j *TrainingJob) ClusterSpec() ClusterSpec {
	clusterSpec := make(ClusterSpec)

	for _, p := range j.Replicas {
		replicaNames := make([]string, 0, *p.Spec.Replicas)

		for i := int32(0); i < *p.Spec.Replicas; i++ {
			replicaNames = append(replicaNames, fmt.Sprintf("%v:%v", p.genName(i), *p.Spec.TFPort))
		}

		clusterSpec[strings.ToLower(string(p.Spec.TFReplicaType))] = replicaNames
	}<commit_msg>Fix a log function use issue (#724)<commit_after>// Copyright 2018 The Kubeflow 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 trainer is to manage TensorFlow training jobs.
package trainer

import (
	"fmt"
	"reflect"
	"strings"

	log "github.com/sirupsen/logrus"
	"k8s.io/api/core/v1"
	"k8s.io/api/policy/v1beta1"
	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/record"

	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/helper"
	tfv1alpha1 "github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1alpha1"
	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/validation"
	tfjobclient "github.com/kubeflow/tf-operator/pkg/client/clientset/versioned"
	"github.com/kubeflow/tf-operator/pkg/client/clientset/versioned/scheme"
	"github.com/kubeflow/tf-operator/pkg/util"
	train_util "github.com/kubeflow/tf-operator/pkg/util/train"
)

// TODO(jlewi): We should switch a New pattern and make trainingJob private so we can
// TrainingJob represents a training job specification.
// ensure correctness on creation.
type TrainingJob struct {
	job *tfv1alpha1.TFJob

	KubeCli kubernetes.Interface

	recorder record.EventRecorder

	Replicas []*TFReplicaSet

	tfJobClient tfjobclient.Interface

	// in memory state of the job.
	// status is the source of truth after job struct is materialized. Changes to the status to be persisted
	// should be made here.
	status tfv1alpha1.TFJobStatus

	memberCounter int

	pdb *v1beta1.PodDisruptionBudget

	// contextLogger is a logger to use for logging information about this replica.
	contextLogger *log.Entry
}

// ClusterSpec represents a cluster TensorFlow specification.
// https://www.tensorflow.org/deploy/distributed#create_a_tftrainclusterspec_to_describe_the_cluster
// It is a map from job names to network addresses.
type ClusterSpec map[string][]string

// TaskSpec represents a Task specification.
type TaskSpec struct {
	Type  string `json:"type"`
	Index int    `json:"index"`
}

//initJob initiate a training job and returns the job specifications.
func initJob(kubeCli kubernetes.Interface, tfJobClient tfjobclient.Interface, recorder record.EventRecorder, job *tfv1alpha1.TFJob) (*TrainingJob, error) {
	j := &TrainingJob{
		KubeCli:     kubeCli,
		tfJobClient: tfJobClient,
		recorder:    recorder,
		Replicas:    make([]*TFReplicaSet, 0),
		job:         job,
		status:      *job.Status.DeepCopy(),

		contextLogger: log.WithFields(log.Fields{
			// We use job to match the key used in controller.go
			// In controller.go we log the key used with the workqueue.
			"job": job.ObjectMeta.Namespace + "/" + job.ObjectMeta.Name,
			"uid": job.ObjectMeta.UID,
		}),
	}

	return j, nil
}

// NewJob method invokes the initJob and check for error
func NewJob(kubeCli kubernetes.Interface, tfJobClient tfjobclient.Interface, recorder record.EventRecorder, job *tfv1alpha1.TFJob, config *tfv1alpha1.ControllerConfig) (*TrainingJob, error) {
	j, err := initJob(kubeCli, tfJobClient, recorder, job)
	if err != nil {
		return nil, err
	}

	return j, nil
}

// Update replaces the TFJob corresponding to TrainingJob with the provided job.
// This function is used when the Spec/Status of the job is modified outside the controller.
// For example, if the user issues a delete request. This will update the metadata on the object
// so we need to replace the spec.
func (j *TrainingJob) Update(newJob *tfv1alpha1.TFJob) {
	j.contextLogger.Infof("Updating job to %+v", *newJob)
	j.job = newJob
}

// UID returns the user ID of the requesting user
func (j *TrainingJob) UID() types.UID {
	return j.job.ObjectMeta.UID
}

// ClusterSpec returns the cluster specification for the training job provided
func (j *TrainingJob) ClusterSpec() ClusterSpec {
	clusterSpec := make(ClusterSpec)

	for _, p := range j.Replicas {
		replicaNames := make([]string, 0, *p.Spec.Replicas)

		for i := int32(0); i < *p.Spec.Replicas; i++ {
			replicaNames = append(replicaNames, fmt.Sprintf("%v:%v", p.genName(i), *p.Spec.TFPort))
		}

		clusterSpec[strings.ToLower(string(p.Spec.TFReplicaType))] = replicaNames
	}
{'fraction_non_alphanumeric': 0.10624801776086266, 'fraction_numerical': 0.009831906121154456, 'mean_word_length': 4.1107509454349005, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 4, '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': '4590512', 'n_tokens_mistral': 3219, 'n_tokens_neox': 3064, 'n_words': 922}