support submit hadoop job dirctly

Author: chriskery

main.go

@@ -18,8 +18,10 @@ package main
 
 import (
 	"flag"
-	"github.com/chriskery/hadoop-cluster-operator/pkg/config"
+	"fmt"
 	"os"
+
+	"github.com/chriskery/hadoop-cluster-operator/pkg/config"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
@@ -55,6 +57,7 @@ func main() {
 	var enableLeaderElection bool
 	var probeAddr string
 	var certDir string
+	var enabledSchemes controllers.EnabledSchemes
 
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
@@ -66,6 +69,8 @@ func main() {
 		config.HadoopInitContainerImageDefault, "The image for hadoop init container")
 	flag.StringVar(&config.Config.HadoopInitContainerTemplateFile, "hadoop-init-container-template-file",
 		config.HadoopInitContainerTemplateFileDefault, "The template file for hadoop init container")
+	flag.Var(&enabledSchemes, "enable-scheme", "Enable scheme(s) as --enable-scheme=tfjob --enable-scheme=pytorchjob, case insensitive."+
+		" Now supporting TFJob, PyTorchJob, MXNetJob, XGBoostJob, PaddleJob. By default, all supported schemes will be enabled.")
 
 	opts := zap.Options{
 		Development: true,
@@ -99,13 +104,22 @@ func main() {
 		os.Exit(1)
 	}
 
-	if err = controllers.NewReconciler(mgr).SetupWithManager(mgr); err != nil {
-		setupLog.Error(err, "unable to create controller", "controller", "hadoopCluster")
-		os.Exit(1)
+	// TODO: We need a general manager. all rest reconciler addsToManager
+	// Based on the user configuration, we start different controllers
+	if enabledSchemes.Empty() {
+		enabledSchemes.FillAll()
 	}
-	if err = (&hadoopclusterorgv1alpha1.HadoopCluster{}).SetupWebhookWithManager(mgr); err != nil {
-		setupLog.Error(err, "unable to create webhook", "webhook", "hadoopCluster")
-		os.Exit(1)
+	for _, s := range enabledSchemes {
+		setupFunc, supported := controllers.SupportedSchemeReconciler[s]
+		if !supported {
+			setupLog.Error(fmt.Errorf("cannot find %s in supportedSchemeReconciler", s),
+				"scheme not supported", "scheme", s)
+			os.Exit(1)
+		}
+		if err = setupFunc(mgr); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", s)
+			os.Exit(1)
+		}
 	}
 	//+kubebuilder:scaffold:builder
 

manifests/crd/bases/kubecluster.org_hadoopjobs.yaml

@@ -3,6 +3,7 @@
 # It should be run by manifests/default
 resources:
 - bases/kubecluster.org_hadoopclusters.yaml
+- bases/kubecluster.org_hadoopjobs.yaml
 #+kubebuilder:scaffold:crdkustomizeresource
 
 patchesStrategicMerge:

manifests/crd/kustomization.yaml

@@ -95,3 +95,29 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - kubecluster.org
+  resources:
+  - hadoopjobs
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - kubecluster.org
+  resources:
+  - hadoopjobs/finalizers
+  verbs:
+  - update
+- apiGroups:
+  - kubecluster.org
+  resources:
+  - hadoopjobs/status
+  verbs:
+  - get
+  - patch
+  - update

manifests/rbac/role.yaml

@@ -0,0 +1,28 @@
+apiVersion: kubecluster.org/v1alpha1
+kind: HadoopJob
+metadata:
+  name: hadoopjob-sample
+spec:
+#  // MainFile is the path to a bundled JAR, Python, or R file of the application.
+#  MainApplicationFile string `json:"mainApplicationFile"`
+#
+#  // Arguments is a list of arguments to be passed to the application.
+#  // +optional
+#  Arguments []string `json:"arguments,omitempty"`
+#
+#  ExecutorSpec HadoopNodeSpec `json:"executorSpec,omitempty"`
+  mainApplicationFile: /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.1.jar
+  arguments: ["pi","10","1000"]
+  executorSpec:
+      image: apache/hadoop:3
+      replicas: 1
+      resources:
+        requests:
+            cpu: 1
+            memory: 4Gi
+        limits:
+            cpu: 1
+            memory: 4Gi
+
+
+

manifests/samples/hadoop_job.yaml

@@ -24,6 +24,26 @@ webhooks:
     resources:
     - hadoopclusters
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /mutate-kubecluster-org-v1alpha1-hadoopjob
+  failurePolicy: Fail
+  name: mhadoopjob.kb.io
+  rules:
+  - apiGroups:
+    - kubecluster.org
+    apiVersions:
+    - v1alpha1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - hadoopjobs
+  sideEffects: None
 ---
 apiVersion: admissionregistration.k8s.io/v1
 kind: ValidatingWebhookConfiguration
@@ -50,3 +70,23 @@ webhooks:
     resources:
     - hadoopclusters
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /validate-kubecluster-org-v1alpha1-hadoopjob
+  failurePolicy: Fail
+  name: vhadoopjob.kb.io
+  rules:
+  - apiGroups:
+    - kubecluster.org
+    apiVersions:
+    - v1alpha1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - hadoopjobs
+  sideEffects: None

manifests/webhook/manifests.yaml

@@ -9,4 +9,6 @@ const (
 	ReplicaTypeNodemanager     ReplicaType = "nodemanager"
 
 	ReplicaTypeConfigMap ReplicaType = "configmap"
+
+	ReplicaTypeDriver ReplicaType = "driver"
 )

pkg/.DS_Store

@@ -0,0 +1,156 @@
+/*
+Copyright 2023.
+
+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 v1alpha1
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	// HadoopJobKind is the kind name.
+	HadoopJobKind = "HadoopJob"
+	// HadoopJobPlural is the TensorflowPlural for HadoopJob.
+	HadoopJobPlural = "HadoopJobs"
+	// HadoopJobSingular is the singular for HadoopJob.
+	HadoopJobSingular = "HadoopJob"
+
+	// JobNameLabel represents the label key for the cluster name, the value is the cluster name.
+	JobNameLabel = "kubeclusetr.org/job-name"
+)
+
+// SparkApplicationType describes the type of a Spark application.
+type SparkApplicationType string
+
+// Different types of Spark applications.
+const (
+	JavaApplicationType   SparkApplicationType = "Java"
+	ScalaApplicationType  SparkApplicationType = "Scala"
+	PythonApplicationType SparkApplicationType = "Python"
+	RApplicationType      SparkApplicationType = "R"
+)
+
+// HadoopJobSpec defines the desired state of HadoopJob
+// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
+type HadoopJobSpec struct {
+	// MainFile is the path to a bundled JAR, Python, or R file of the application.
+	MainApplicationFile string `json:"mainApplicationFile"`
+
+	// Arguments is a list of arguments to be passed to the application.
+	// +optional
+	Arguments []string `json:"arguments,omitempty"`
+
+	ExecutorSpec HadoopNodeSpec `json:"executorSpec,omitempty"`
+}
+
+// +k8s:openapi-gen=true
+// +k8s:deepcopy-gen=true
+// JobCondition describes current state of a cluster
+type JobCondition struct {
+	// Type of job condition.
+	Type JobConditionType `json:"type"`
+	// Status of the condition, one of True, False, Unknown.
+	Status corev1.ConditionStatus `json:"status"`
+	// The reason for the condition's last transition.
+	Reason string `json:"reason,omitempty"`
+	// A human readable message indicating details about the transition.
+	Message string `json:"message,omitempty"`
+	// The last time this condition was updated.
+	LastUpdateTime metav1.Time `json:"lastUpdateTime,omitempty"`
+	// Last time the condition transitioned from one status to another.
+	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty"`
+}
+
+type JobConditionType string
+
+const (
+	// JobCreated means the job has been accepted by the system,
+	// but one or more of the pods/services has not been started.
+	// This includes time before pods being scheduled and launched.
+	JobCreated JobConditionType = "Created"
+
+	// JobSubmitted means all sub-resources (e.g. services/pods) of this job
+	// have been successfully submitted.
+	JobSubmitted JobConditionType = "Submitted"
+
+	// JobRunning means all sub-resources (e.g. services/pods) of this job
+	// have been successfully scheduled and launched.
+	// The training is running without error.
+	JobRunning JobConditionType = "Running"
+
+	// JobSucceeded means all sub-resources (e.g. services/pods) of this job
+	// reached phase have terminated in success.
+	// The training is complete without error.
+	JobSucceeded JobConditionType = "Succeeded"
+
+	// JobFailed means one or more sub-resources (e.g. services/pods) of this job
+	// reached phase failed with no restarting.
+	// The training has failed its execution.
+	JobFailed JobConditionType = "Failed"
+)
+
+// +k8s:openapi-gen=true
+// +k8s:deepcopy-gen=true
+// HadoopJobStatus defines the observed state of HadoopJob
+type HadoopJobStatus struct {
+	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
+	// Important: Run "make" to regenerate code after modifying this file
+	// Conditions is an array of current observed job conditions.
+	Conditions []JobCondition `json:"conditions"`
+
+	// Represents time when the job was acknowledged by the job controller.
+	// It is not guaranteed to be set in happens-before order across separate operations.
+	// It is represented in RFC3339 form and is in UTC.
+	StartTime *metav1.Time `json:"startTime,omitempty"`
+
+	// Represents time when the job was completed. It is not guaranteed to
+	// be set in happens-before order across separate operations.
+	// It is represented in RFC3339 form and is in UTC.
+	CompletionTime *metav1.Time `json:"completionTime,omitempty"`
+}
+
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +resource:path=hadoopjobs
+// +kubebuilder:object:root=true
+// +kubebuilder:printcolumn:JSONPath=`.metadata.creationTimestamp`,name="Age",type=date
+// +kubebuilder:printcolumn:JSONPath=`.status.conditions[-1:].type`,name="State",type=string
+// +kubebuilder:subresource:status
+// +kubebuilder:resource:scope=Namespaced,path=hadoopjobs,shortName={"hdj","hdjs"}
+// HadoopJob is the Schema for the hadoopjobs API
+type HadoopJob struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   HadoopJobSpec   `json:"spec,omitempty"`
+	Status HadoopJobStatus `json:"status,omitempty"`
+}
+
+//+kubebuilder:object:root=true
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimzxachinery/pkg/runtime.Object
+// +resource:path=hadoopjobs
+
+// HadoopJobList contains a list of HadoopJob
+type HadoopJobList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []HadoopJob `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&HadoopJob{}, &HadoopJobList{})
+}