minor chnages

Author: Nathan-Salawe

src/cmd/create/es.go

@@ -17,6 +17,7 @@ var (
 
 	ElasticSearchVersion string = "8.15.0"
 	Nodes                uint16 = 1
+	esNamespace          string = "default"
 )
 
 // infoCmd represents the info command
@@ -38,8 +39,8 @@ var ElasticSearchCmd = &cobra.Command{
 				{
 					Name:  args[0],
 					Count: Nodes,
-					Config: types.EsNodeConfig{
-						NodeStoreAllowMMAP: false,
+					Config: map[string]interface{}{
+						"node.store.allow_mmap": false,
 					},
 				},
 			},
@@ -51,8 +52,8 @@ var ElasticSearchCmd = &cobra.Command{
 				Kind:       "Elasticsearch",
 			},
 			metav1.ObjectMeta{
-				Name:      "default",
-				Namespace: "elastic-system",
+				Name:      args[0],
+				Namespace: esNamespace,
 			},
 			spec,
 			"elasticsearches",
@@ -62,12 +63,13 @@ var ElasticSearchCmd = &cobra.Command{
 			fmt.Println(err)
 		}
 		fmt.Println("Elastic search created")
-		fmt.Printf("incluster url: http://%s.default.svc.cluster.local:9200 \n", args[0])
-		fmt.Printf("Es Password can be retrieved using kubectl \n kubectl get secret %s-es-elastic-user -o go-template='{{.data.elastic | base64decode}}'\n", args[0])
+		fmt.Printf("incluster url: http://%s-es-http.default.svc.cluster.local:9200 \n", args[0])
+		fmt.Printf("Elasticsearch Password can be retrieved using kubectl \n kubectl get secret %s-es-elastic-user -n %s -o go-template='{{.data.elastic | base64decode}}'\n", args[0], esNamespace)
 	},
 }
 
 func init() {
 	ElasticSearchCmd.Flags().StringVar(&ElasticSearchVersion, "version", ElasticSearchVersion, "elastic search version")
 	ElasticSearchCmd.Flags().Uint16Var(&Nodes, "nodes", Nodes, "number of elastic search Nodes")
+	ElasticSearchCmd.Flags().StringVar(&esNamespace, "namespace", esNamespace, "namespace to deploy create elasticsearch cluster on")
 }

src/cmd/create/flink.go

@@ -15,9 +15,9 @@ import (
 
 var (
 	cpu         string = "500m"
-	memory      string = "2048Mi"
+	memory      string = "1Gi"
 	cpuLimit    string = "1"
-	memoryLimit string = "4096Mi"
+	memoryLimit string = "2Gi"
 	volumeSize  string = "1Gi"
 	taskslots   uint8  = 1
 	replicas    uint8  = 1
@@ -55,7 +55,8 @@ var FlinkClusterCmd = &cobra.Command{
 		}
 		namespace := "flink"
 
-		flinkVersion := "v1_16"
+		flinkVersion := "1.16"
+		flinkVersionLong := "v1_16"
 		ClaimName := fmt.Sprintf("%s-pvc", args[0])
 		fmt.Printf("creating flink cluster %s\n", args[0])
 
@@ -73,15 +74,16 @@ var FlinkClusterCmd = &cobra.Command{
 		}
 
 		var spec types.FlinkDeploymentSpec
-		taskmanagerImage := fmt.Sprintf("beamstackproj/beam-harness-%s:latest", flinkVersion)
+		workerImage := "beamstackproj/beam-harness:latest"
 
 		if Previledged {
-			flinkImage := fmt.Sprintf("beamstackproj/flink-%s-docker:latest", flinkVersion)
+			// flinkImage := fmt.Sprintf("beamstackproj/flink-%s-docker:latest", flinkVersion)
+			flinkImage := fmt.Sprintf("flink:%s", flinkVersion)
 
 			spec = types.FlinkDeploymentSpec{
 				Image:           &flinkImage,
 				ImagePullPolicy: "IfNotPresent",
-				FlinkVersion:    flinkVersion,
+				FlinkVersion:    flinkVersionLong,
 				FlinkConfiguration: map[string]string{
 					"taskmanager.numberOfTaskSlots": fmt.Sprintf("%d", taskslots),
 				},
@@ -124,7 +126,7 @@ var FlinkClusterCmd = &cobra.Command{
 							Containers: []v1.Container{
 								{
 									Name:  "worker",
-									Image: taskmanagerImage,
+									Image: workerImage,
 									Args:  []string{"-worker_pool"},
 									Ports: []v1.ContainerPort{
 										{
@@ -179,11 +181,12 @@ var FlinkClusterCmd = &cobra.Command{
 				},
 			}
 		} else {
-			flinkImage := fmt.Sprintf("beamstackproj/flink-%s:latest", flinkVersion)
+			flinkImage := fmt.Sprintf("beamstackproj/flink-%s:latest", flinkVersionLong)
+			// flinkImage := fmt.Sprintf("flink:%s", flinkVersion)
 			spec = types.FlinkDeploymentSpec{
 				Image:           &flinkImage,
 				ImagePullPolicy: "IfNotPresent",
-				FlinkVersion:    flinkVersion,
+				FlinkVersion:    flinkVersionLong,
 				FlinkConfiguration: map[string]string{
 					"taskmanager.numberOfTaskSlots": fmt.Sprintf("%d", taskslots),
 				},
@@ -231,7 +234,7 @@ var FlinkClusterCmd = &cobra.Command{
 							Containers: []v1.Container{
 								{
 									Name:  "worker",
-									Image: taskmanagerImage,
+									Image: workerImage,
 									Args:  []string{"-worker_pool"},
 									Ports: []v1.ContainerPort{
 										{
@@ -294,5 +297,8 @@ func init() {
 	FlinkClusterCmd.Flags().Uint8Var(&replicas, "replicas", replicas, "numbers of replicas sets for task manager")
 	FlinkClusterCmd.Flags().Uint8Var(&taskslots, "taskslots", taskslots, "numbers of taskslots to be created for the task manager")
 	FlinkClusterCmd.Flags().StringVar(&volumeSize, "volumeSize", volumeSize, "size of persistent volume to be attached to flink cluster")
-	FlinkClusterCmd.Flags().BoolVar(&Previledged, "Previledged", Previledged, "")
+	FlinkClusterCmd.Flags().BoolVarP(&Previledged, "previledged", "p", Previledged, "")
 }
+
+// export ELASTIC_PASSWORD="admin"
+// export KIBANA_PASSWORD="kibana-pass"

src/cmd/deploy/pipeline.go

@@ -290,8 +290,7 @@ func DeployPipeline(cmd *cobra.Command, args []string) {
 						Containers: []v1.Container{
 							{
 								Name:  "beam-pipeline",
-								Image: "beamstackproj/beam-harness-v1_16:latest",
-								// Image:   "localhost:5000/docker-ext-v1:latest",
+								Image: "beamstackproj/beam-harness:latest",
 								Command: []string{"python"},
 								Args: []string{
 									"-m",

src/cmd/initialize/init.go

@@ -33,7 +33,8 @@ var Banner string = `
  ____)  ) _____  _____  ____    ___  _| |_  _____   ____ | |  _ 
 |  __  ( | ___ |(____ ||    \  /___)(_   _)(____ | / ___)| |_/ )
 | |__)  )| ____|/ ___ || | | ||___ |  | |_ / ___ |( (___ |  _ ( 
-|______/ |_____)\_____||_|_|_|(___/    \__)\_____| \____)|_| \_)                                                                
+|______/ |_____)\_____||_|_|_|(___/    \__)\_____| \____)|_| \_)      
+
 `
 
 var (
@@ -58,8 +59,8 @@ func init() {
 	InitCmd.Flags().StringVarP(&Name, "name", "n", Name, "Name of profile. will be randomly generated if not provided.")
 	InitCmd.Flags().StringVarP(&ConfigFile, "config", "c", ConfigFile, "Path to configuration file.")
 	InitCmd.Flags().StringVarP(&DefaultOperator, "default-operator", "d", DefaultOperator, "Default operator.")
-	InitCmd.Flags().StringVarP(&FlinkVersion, "flink-version", "f", FlinkVersion, "Flink version to be installed. Ignored if Flink is not specified for installation.")
-	InitCmd.Flags().StringVarP(&SparkVersion, "spark-version", "s", SparkVersion, "Spark Version to be installed. Ignored if Spark is not specified for installation.")
+	InitCmd.Flags().StringVarP(&FlinkVersion, "flink-version", "f", FlinkVersion, "Flink kubernetes operator version to be installed. Ignored if Flink is not specified for installation.")
+	InitCmd.Flags().StringVarP(&SparkVersion, "spark-version", "s", SparkVersion, "Spark kubernetes operator Version to be installed. Ignored if Spark is not specified for installation.")
 	InitCmd.Flags().BoolVarP(&Flink, "flink", "F", Flink, "If specified, flink is installed.")
 	InitCmd.Flags().BoolVarP(&Spark, "spark", "S", Spark, "If specified, Spark is installed.")
 	InitCmd.Flags().BoolVarP(&force, "force", "q", force, "If specified, will automatically reinitialize cluster")
@@ -68,8 +69,6 @@ func init() {
 }
 
 func runInit(cmd *cobra.Command, args []string) {
-	// fmt.Println("Initializing cluster ! !")
-	// out, _ := glamour.Render(fmt.Sprintf("# %s", Banner), "dark")
 	fmt.Print(Banner)
 	currentContext, err := utils.GetCurrentContext()
 	if err != nil {

src/types/es.go

@@ -1,13 +1,9 @@
 package types
 
-type EsNodeConfig struct {
-	NodeStoreAllowMMAP bool `yaml:"node.store.allow_mmap"`
-}
-
 type EsNodeSet struct {
-	Name   string       `yaml:"name"`
-	Count  uint16       `yaml:"count"`
-	Config EsNodeConfig `yaml:"config"`
+	Name   string      `yaml:"name"`
+	Count  uint16      `yaml:"count"`
+	Config map[string]interface{} `yaml:"config"`
 }
 
 type EsSpec struct {