Skip to content

Commit 89b24c1

Browse files
Nathan-SalaweNathan-Salawe
authored
Merge pull request #36 from BeamStackProj/release
Release
2 parents 31ee44a + 0aa991a commit 89b24c1

File tree

13 files changed

+392
-137
lines changed

13 files changed

+392
-137
lines changed

‎releases/latest/beamstack-linux-amd64

89.9 MB
Binary file not shown.

‎src/cmd/create/create.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,12 @@ import (
1010
// infoCmd represents the info command
1111
var CreateCmd = &cobra.Command{
1212
Use: "create",
13-
Short: "'create' sub commands",
13+
Short: "create a resource",
1414
Long: `create flink or spark clusters`,
1515
}
1616

1717
func init() {
1818

1919
CreateCmd.AddCommand(FlinkClusterCmd)
20+
CreateCmd.AddCommand(ElasticSearchCmd)
2021
}

‎src/cmd/create/es.go

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package create
2+
3+
import (
4+
"fmt"
5+
6+
"github.com/BeamStackProj/beamstack-cli/src/objects"
7+
"github.com/BeamStackProj/beamstack-cli/src/types"
8+
"github.com/BeamStackProj/beamstack-cli/src/utils"
9+
"github.com/spf13/cobra"
10+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
11+
)
12+
13+
var (
14+
esLongDesc = utils.LongDesc(`
15+
Create a elastic search with specified requirments.
16+
`)
17+
18+
ElasticSearchVersion string = "8.15.0"
19+
Nodes uint16 = 1
20+
)
21+
22+
// infoCmd represents the info command
23+
var ElasticSearchCmd = &cobra.Command{
24+
Use: "elasticsearch [NAME]",
25+
Short: "create elasticsearch cluster",
26+
Long: esLongDesc,
27+
Args: func(cmd *cobra.Command, args []string) error {
28+
if len(args) != 1 {
29+
return fmt.Errorf("es command requires exactly one argument: cluster Name. Provided %d arguments", len(args))
30+
}
31+
return nil
32+
},
33+
Run: func(cmd *cobra.Command, args []string) {
34+
35+
spec := types.EsSpec{
36+
Version: ElasticSearchVersion,
37+
NodeSets: []types.EsNodeSet{
38+
{
39+
Name: args[0],
40+
Count: Nodes,
41+
Config: types.EsNodeConfig{
42+
NodeStoreAllowMMAP: false,
43+
},
44+
},
45+
},
46+
}
47+
48+
err := objects.CreateDynamicResource(
49+
metav1.TypeMeta{
50+
APIVersion: "elasticsearch.k8s.elastic.co/v1",
51+
Kind: "Elasticsearch",
52+
},
53+
metav1.ObjectMeta{
54+
Name: "default",
55+
Namespace: "elastic-system",
56+
},
57+
spec,
58+
"elasticsearches",
59+
)
60+
61+
if err != nil {
62+
fmt.Println(err)
63+
}
64+
fmt.Println("Elastic search created")
65+
fmt.Printf("incluster url: http://%s.default.svc.cluster.local:9200 \n", args[0])
66+
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])
67+
},
68+
}
69+
70+
func init() {
71+
ElasticSearchCmd.Flags().StringVar(&ElasticSearchVersion, "version", ElasticSearchVersion, "elastic search version")
72+
ElasticSearchCmd.Flags().Uint16Var(&Nodes, "nodes", Nodes, "number of elastic search Nodes")
73+
}

‎src/cmd/create/flink.go

Lines changed: 167 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,14 @@ import (
1414
)
1515

1616
var (
17-
cpu string = "2"
17+
cpu string = "4"
1818
memory string = "2048Mi"
19-
cpuLimit string = "2"
19+
cpuLimit string = "4"
2020
memoryLimit string = "2048Mi"
2121
volumeSize string = "1Gi"
2222
taskslots uint8 = 10
2323
replicas uint8 = 1
24+
Previledged bool = false
2425
)
2526

2627
// Description and Examples for creating flink clsuters
@@ -35,7 +36,12 @@ var FlinkClusterCmd = &cobra.Command{
3536
Use: "flink [NAME]",
3637
Short: "create a flink cluster",
3738
Long: flinkLongDesc,
38-
Args: cobra.ExactArgs(1),
39+
Args: func(cmd *cobra.Command, args []string) error {
40+
if len(args) != 1 {
41+
return fmt.Errorf("flink command requires exactly one argument: cluster Name. Provided %d arguments", len(args))
42+
}
43+
return nil
44+
},
3945
Run: func(cmd *cobra.Command, args []string) {
4046
profile, err := utils.ValidateCluster()
4147

@@ -47,11 +53,9 @@ var FlinkClusterCmd = &cobra.Command{
4753
fmt.Println("Flink Operator not initialized on this cluster")
4854
return
4955
}
50-
5156
namespace := "flink"
57+
5258
flinkVersion := "v1_16"
53-
flinkImage := fmt.Sprintf("beamstackproj/flink-%s:latest", flinkVersion)
54-
taskmanagerImage := fmt.Sprintf("beamstackproj/beam-harness-%s:latest", flinkVersion)
5559
ClaimName := fmt.Sprintf("%s-pvc", args[0])
5660
fmt.Printf("creating flink cluster %s\n", args[0])
5761

@@ -68,82 +72,188 @@ var FlinkClusterCmd = &cobra.Command{
6872
return
6973
}
7074

71-
spec := types.FlinkDeploymentSpec{
72-
Image: &flinkImage,
73-
ImagePullPolicy: "IfNotPresent",
74-
FlinkVersion: flinkVersion,
75-
FlinkConfiguration: map[string]string{
76-
"taskmanager.numberOfTaskSlots": fmt.Sprintf("%d", taskslots),
77-
},
78-
ServiceAccount: "flink",
79-
PodTemplate: &v1.PodTemplateSpec{
80-
Spec: v1.PodSpec{
81-
Containers: []v1.Container{
82-
{
83-
Name: "flink-main-container",
84-
VolumeMounts: []v1.VolumeMount{
85-
{
86-
MountPath: "/opt/flink/log",
87-
Name: "flink-logs",
75+
var spec types.FlinkDeploymentSpec
76+
taskmanagerImage := fmt.Sprintf("beamstackproj/beam-harness-%s:latest", flinkVersion)
77+
78+
if Previledged {
79+
flinkImage := fmt.Sprintf("beamstackproj/flink-%s-docker:latest", flinkVersion)
80+
81+
spec = types.FlinkDeploymentSpec{
82+
Image: &flinkImage,
83+
ImagePullPolicy: "IfNotPresent",
84+
FlinkVersion: flinkVersion,
85+
FlinkConfiguration: map[string]string{
86+
"taskmanager.numberOfTaskSlots": fmt.Sprintf("%d", taskslots),
87+
},
88+
ServiceAccount: "flink",
89+
PodTemplate: &v1.PodTemplateSpec{
90+
Spec: v1.PodSpec{
91+
Containers: []v1.Container{
92+
{
93+
Name: "flink-main-container",
94+
Image: flinkImage,
95+
VolumeMounts: []v1.VolumeMount{},
96+
SecurityContext: &v1.SecurityContext{
97+
Privileged: func(b bool) *bool { return &b }(true),
8898
},
8999
},
90100
},
101+
Volumes: []v1.Volume{},
91102
},
92-
Volumes: []v1.Volume{
93-
{
94-
Name: "flink-logs",
95-
},
103+
},
104+
JobManager: types.JobManagerSpec{
105+
Replicas: 1,
106+
Resource: types.Resource{
107+
Memory: memory,
108+
CPU: cpu,
96109
},
97110
},
98-
},
99-
JobManager: types.JobManagerSpec{
100-
Replicas: 1,
101-
Resource: types.Resource{
102-
Memory: memory,
103-
CPU: cpu,
111+
TaskManager: types.TaskManagerSpec{
112+
Replicas: replicas,
113+
Resource: types.Resource{
114+
Memory: memory,
115+
CPU: cpu,
116+
},
117+
118+
PodTemplate: &v1.PodTemplateSpec{
119+
Spec: v1.PodSpec{
120+
Containers: []v1.Container{
121+
{
122+
Name: "worker",
123+
Image: taskmanagerImage,
124+
Args: []string{"-worker_pool"},
125+
Ports: []v1.ContainerPort{
126+
{
127+
Name: "harness-port",
128+
ContainerPort: 50000,
129+
},
130+
},
131+
VolumeMounts: []v1.VolumeMount{
132+
{
133+
MountPath: "/pvc",
134+
Name: "flink-cluster-pvc",
135+
},
136+
},
137+
},
138+
{
139+
Name: "flink-main-container",
140+
SecurityContext: &v1.SecurityContext{
141+
Privileged: func(b bool) *bool { return &b }(true),
142+
},
143+
VolumeMounts: []v1.VolumeMount{
144+
{
145+
MountPath: "/var/run/docker.sock",
146+
Name: "docker-socket",
147+
},
148+
},
149+
},
150+
},
151+
Volumes: []v1.Volume{
152+
{
153+
Name: "flink-cluster-pvc",
154+
VolumeSource: v1.VolumeSource{
155+
PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
156+
ClaimName: ClaimName,
157+
},
158+
},
159+
},
160+
{
161+
Name: "docker-socket",
162+
VolumeSource: v1.VolumeSource{
163+
HostPath: &v1.HostPathVolumeSource{
164+
Path: "/var/run/docker.sock",
165+
Type: func() *v1.HostPathType {
166+
t := v1.HostPathSocket
167+
return &t
168+
}(),
169+
},
170+
},
171+
},
172+
},
173+
},
174+
},
104175
},
105-
},
106-
TaskManager: types.TaskManagerSpec{
107-
Replicas: replicas,
108-
Resource: types.Resource{
109-
Memory: memory,
110-
CPU: cpu,
176+
}
177+
} else {
178+
flinkImage := fmt.Sprintf("beamstackproj/flink-%s:latest", flinkVersion)
179+
spec = types.FlinkDeploymentSpec{
180+
Image: &flinkImage,
181+
ImagePullPolicy: "IfNotPresent",
182+
FlinkVersion: flinkVersion,
183+
FlinkConfiguration: map[string]string{
184+
"taskmanager.numberOfTaskSlots": fmt.Sprintf("%d", taskslots),
111185
},
112-
186+
ServiceAccount: "flink",
113187
PodTemplate: &v1.PodTemplateSpec{
114188
Spec: v1.PodSpec{
115189
Containers: []v1.Container{
116190
{
117-
Name: "worker",
118-
Image: taskmanagerImage,
119-
Args: []string{"-worker_pool"},
120-
Ports: []v1.ContainerPort{
121-
{
122-
Name: "harness-port",
123-
ContainerPort: 50000,
124-
},
125-
},
191+
Name: "flink-main-container",
126192
VolumeMounts: []v1.VolumeMount{
127193
{
128-
MountPath: "/pvc",
129-
Name: "flink-cluster-pvc",
194+
MountPath: "/opt/flink/log",
195+
Name: "flink-logs",
130196
},
131197
},
132198
},
133199
},
134200
Volumes: []v1.Volume{
135201
{
136-
Name: "flink-cluster-pvc",
137-
VolumeSource: v1.VolumeSource{
138-
PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
139-
ClaimName: ClaimName,
202+
Name: "flink-logs",
203+
},
204+
},
205+
},
206+
},
207+
JobManager: types.JobManagerSpec{
208+
Replicas: 1,
209+
Resource: types.Resource{
210+
Memory: memory,
211+
CPU: cpu,
212+
},
213+
},
214+
TaskManager: types.TaskManagerSpec{
215+
Replicas: replicas,
216+
Resource: types.Resource{
217+
Memory: memory,
218+
CPU: cpu,
219+
},
220+
221+
PodTemplate: &v1.PodTemplateSpec{
222+
Spec: v1.PodSpec{
223+
Containers: []v1.Container{
224+
{
225+
Name: "worker",
226+
Image: taskmanagerImage,
227+
Args: []string{"-worker_pool"},
228+
Ports: []v1.ContainerPort{
229+
{
230+
Name: "harness-port",
231+
ContainerPort: 50000,
232+
},
233+
},
234+
VolumeMounts: []v1.VolumeMount{
235+
{
236+
MountPath: "/pvc",
237+
Name: "flink-cluster-pvc",
238+
},
239+
},
240+
},
241+
},
242+
Volumes: []v1.Volume{
243+
{
244+
Name: "flink-cluster-pvc",
245+
VolumeSource: v1.VolumeSource{
246+
PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
247+
ClaimName: ClaimName,
248+
},
140249
},
141250
},
142251
},
143252
},
144253
},
145254
},
146-
},
255+
}
256+
147257
}
148258

149259
err = objects.CreateDynamicResource(
@@ -176,4 +286,5 @@ func init() {
176286
FlinkClusterCmd.Flags().Uint8Var(&replicas, "replicas", replicas, "numbers of replicas sets for task manager")
177287
FlinkClusterCmd.Flags().Uint8Var(&taskslots, "taskslots", taskslots, "numbers of taskslots to be created for the task manager")
178288
FlinkClusterCmd.Flags().StringVar(&volumeSize, "volumeSize", volumeSize, "size of persistent volume to be attached to flink cluster")
289+
FlinkClusterCmd.Flags().BoolVar(&Previledged, "Previledged", Previledged, "")
179290
}

0 commit comments

Comments
 (0)