Skip to content

Commit 738f4f7

Browse files
freeznetfreeznet
committed
make function path consistency when initContainer downloader enabled
1 parent a253638 commit 738f4f7

File tree

5 files changed

+111
-11
lines changed

5 files changed

+111
-11
lines changed

‎controllers/spec/common.go

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1801,14 +1801,16 @@ func generateDownloaderVolumeMountsForDownloader(javaRuntime *v1alpha1.JavaRunti
18011801
}
18021802

18031803
func generateDownloaderVolumeMountsForRuntime(javaRuntime *v1alpha1.JavaRuntime, pythonRuntime *v1alpha1.PythonRuntime,
1804-
goRuntime *v1alpha1.GoRuntime) []corev1.VolumeMount {
1804+
goRuntime *v1alpha1.GoRuntime, genericRuntime *v1alpha1.GenericRuntime) []corev1.VolumeMount {
18051805
downloadPath := ""
18061806
if javaRuntime != nil && javaRuntime.JarLocation != "" {
18071807
downloadPath = javaRuntime.Jar
18081808
} else if pythonRuntime != nil && pythonRuntime.PyLocation != "" {
18091809
downloadPath = pythonRuntime.Py
18101810
} else if goRuntime != nil && goRuntime.GoLocation != "" {
18111811
downloadPath = goRuntime.Go
1812+
} else if genericRuntime != nil && genericRuntime.FunctionFile != "" {
1813+
downloadPath = genericRuntime.FunctionFile
18121814
}
18131815

18141816
if downloadPath != "" {
@@ -1830,9 +1832,10 @@ func generateDownloaderVolumeMountsForRuntime(javaRuntime *v1alpha1.JavaRuntime,
18301832
SubPath: subPath,
18311833
}}
18321834
}
1835+
idx := strings.LastIndex(mountPath, "/")
18331836
return []corev1.VolumeMount{{
18341837
Name: DownloaderVolume,
1835-
MountPath: mountPath[:len(mountPath)-len(subPath)],
1838+
MountPath: mountPath[:idx],
18361839
}}
18371840
}
18381841
return nil

‎controllers/spec/function.go

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package spec
2020
import (
2121
"context"
2222
"regexp"
23+
"strings"
2324

2425
"github.com/streamnative/function-mesh/api/compute/v1alpha1"
2526
"github.com/streamnative/function-mesh/utils"
@@ -177,7 +178,7 @@ func makeFunctionContainer(function *v1alpha1.Function) *corev1.Container {
177178
mounts := makeFunctionVolumeMounts(function, function.Spec.Pulsar.AuthConfig)
178179
if utils.EnableInitContainers {
179180
mounts = append(mounts,
180-
generateDownloaderVolumeMountsForRuntime(function.Spec.Java, function.Spec.Python, function.Spec.Golang)...)
181+
generateDownloaderVolumeMountsForRuntime(function.Spec.Java, function.Spec.Python, function.Spec.Golang, function.Spec.GenericRuntime)...)
181182
}
182183
return &corev1.Container{
183184
// TODO new container to pull user code image and upload jars into bookkeeper
@@ -230,7 +231,8 @@ func makeFunctionCommand(function *v1alpha1.Function) []string {
230231
}
231232
if spec.Java != nil {
232233
if spec.Java.Jar != "" {
233-
return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
234+
mountPath := extractMountPath(spec.Java.Jar)
235+
return MakeJavaFunctionCommand(spec.Java.JarLocation, mountPath,
234236
spec.Name, spec.ClusterName,
235237
GenerateJavaLogConfigCommand(spec.Java, spec.LogTopicAgent),
236238
parseJavaLogLevel(spec.Java),
@@ -246,7 +248,8 @@ func makeFunctionCommand(function *v1alpha1.Function) []string {
246248
}
247249
} else if spec.Python != nil {
248250
if spec.Python.Py != "" {
249-
return MakePythonFunctionCommand(spec.Python.PyLocation, spec.Python.Py,
251+
mountPath := extractMountPath(spec.Python.Py)
252+
return MakePythonFunctionCommand(spec.Python.PyLocation, mountPath,
250253
spec.Name, spec.ClusterName,
251254
generatePythonLogConfigCommand(spec.Name, spec.Python, spec.LogTopicAgent),
252255
generateFunctionDetailsInJSON(function), string(function.UID), hasPulsarctl, hasWget,
@@ -255,11 +258,13 @@ func makeFunctionCommand(function *v1alpha1.Function) []string {
255258
}
256259
} else if spec.Golang != nil {
257260
if spec.Golang.Go != "" {
258-
return MakeGoFunctionCommand(spec.Golang.GoLocation, spec.Golang.Go, function)
261+
mountPath := extractMountPath(spec.Golang.Go)
262+
return MakeGoFunctionCommand(spec.Golang.GoLocation, mountPath, function)
259263
}
260264
} else if spec.GenericRuntime != nil {
261265
if spec.GenericRuntime.FunctionFile != "" {
262-
return MakeGenericFunctionCommand(spec.GenericRuntime.FunctionFileLocation, spec.GenericRuntime.FunctionFile,
266+
mountPath := extractMountPath(spec.GenericRuntime.FunctionFile)
267+
return MakeGenericFunctionCommand(spec.GenericRuntime.FunctionFileLocation, mountPath,
263268
spec.GenericRuntime.Language, spec.ClusterName,
264269
generateFunctionDetailsInJSON(function), string(function.UID),
265270
spec.Pulsar.AuthSecret != "", spec.Pulsar.TLSSecret != "", function.Spec.SecretsMap,
@@ -280,3 +285,18 @@ func generateFunctionDetailsInJSON(function *v1alpha1.Function) string {
280285
log.Info(string(json))
281286
return string(json)
282287
}
288+
289+
func extractMountPath(p string) string {
290+
if utils.EnableInitContainers {
291+
mountPath := p
292+
// for relative path, volume should be mounted to the WorkDir
293+
// and path also should be under the $WorkDir dir
294+
if !strings.HasPrefix(p, "/") {
295+
mountPath = WorkDir + p
296+
} else if !strings.HasPrefix(p, WorkDir) {
297+
mountPath = strings.Replace(p, "/", WorkDir, 1)
298+
}
299+
return mountPath
300+
}
301+
return p
302+
}

‎controllers/spec/function_test.go

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@ import (
2121
"strings"
2222
"testing"
2323

24+
"github.com/streamnative/function-mesh/utils"
25+
corev1 "k8s.io/api/core/v1"
26+
2427
"github.com/streamnative/function-mesh/api/compute/v1alpha1"
2528

2629
"gotest.tools/assert"
@@ -93,3 +96,54 @@ func makeFunctionSample(functionName string) *v1alpha1.Function {
9396
},
9497
}
9598
}
99+
100+
func makeFunctionSamplePackageUrl(functionName string) *v1alpha1.Function {
101+
f := makeFunctionSample(functionName)
102+
f.Spec.Java.JarLocation = "function://public/default/java-function"
103+
f.Spec.Java.Jar = "/tmp/java-function.jar"
104+
return f
105+
}
106+
107+
func TestInitContainerDownloader(t *testing.T) {
108+
utils.EnableInitContainers = true
109+
function := makeFunctionSamplePackageUrl("test")
110+
111+
objectMeta := MakeFunctionObjectMeta(function)
112+
113+
runnerImagePullSecrets := getFunctionRunnerImagePullSecret()
114+
for _, mapSecret := range runnerImagePullSecrets {
115+
if value, ok := mapSecret["name"]; ok {
116+
function.Spec.Pod.ImagePullSecrets = append(function.Spec.Pod.ImagePullSecrets, corev1.LocalObjectReference{Name: value})
117+
}
118+
}
119+
runnerImagePullPolicy := getFunctionRunnerImagePullPolicy()
120+
function.Spec.ImagePullPolicy = runnerImagePullPolicy
121+
122+
labels := makeFunctionLabels(function)
123+
statefulSet := MakeStatefulSet(objectMeta, function.Spec.Replicas, function.Spec.DownloaderImage,
124+
makeFunctionContainer(function), makeFunctionVolumes(function, function.Spec.Pulsar.AuthConfig), labels, function.Spec.Pod,
125+
function.Spec.Pulsar.AuthConfig, function.Spec.Pulsar.TLSConfig, function.Spec.Pulsar.PulsarConfig, function.Spec.Pulsar.AuthSecret,
126+
function.Spec.Pulsar.TLSSecret, function.Spec.Java, function.Spec.Python, function.Spec.Golang, function.Spec.Pod.Env, function.Name,
127+
function.Spec.LogTopic, function.Spec.FilebeatImage, function.Spec.LogTopicAgent, function.Spec.VolumeMounts,
128+
function.Spec.VolumeClaimTemplates, function.Spec.PersistentVolumeClaimRetentionPolicy)
129+
130+
assert.Assert(t, statefulSet != nil, "statefulSet should not be nil")
131+
assert.Assert(t, len(statefulSet.Spec.Template.Spec.InitContainers) == 1, "init container should be 1 but got %d", len(statefulSet.Spec.Template.Spec.InitContainers))
132+
assert.Assert(t, statefulSet.Spec.Template.Spec.InitContainers[0].Name == "downloader", "init container name should be downloader but got %s", statefulSet.Spec.Template.Spec.InitContainers[0].Name)
133+
downloaderCommands := statefulSet.Spec.Template.Spec.InitContainers[0].Command
134+
functionCommands := makeFunctionCommand(function)
135+
assert.Assert(t, len(downloaderCommands) == 3, "downloader commands should be 3 but got %d", len(downloaderCommands))
136+
assert.Assert(t, len(functionCommands) == 3, "function commands should be 3 but got %d", len(functionCommands))
137+
assert.Assert(t, len(statefulSet.Spec.Template.Spec.InitContainers[0].VolumeMounts) == 1, "volume mounts should be 1 but got %d", len(statefulSet.Spec.Template.Spec.InitContainers[0].VolumeMounts))
138+
assert.Assert(t, len(statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts) == 1, "volume mounts should be 1 but got %d", len(statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts))
139+
startDownloadCommands := downloaderCommands[2]
140+
downloadVolumeMount := statefulSet.Spec.Template.Spec.InitContainers[0].VolumeMounts[0]
141+
assert.Assert(t, downloadVolumeMount.Name == "downloader-volume", "volume mount name should be download-volume but got %s", downloadVolumeMount.Name)
142+
assert.Assert(t, downloadVolumeMount.MountPath == "/pulsar/download", "volume mount path should be /pulsar/download but got %s", downloadVolumeMount.MountPath)
143+
startFunctionCommands := functionCommands[2]
144+
functionVolumeMount := statefulSet.Spec.Template.Spec.Containers[0].VolumeMounts[0]
145+
assert.Assert(t, functionVolumeMount.Name == "downloader-volume", "volume mount name should be downloader-volume but got %s", functionVolumeMount.Name)
146+
assert.Assert(t, functionVolumeMount.MountPath == "/pulsar/tmp", "volume mount path should be /pulsar/tmp but got %s", functionVolumeMount.MountPath)
147+
assert.Assert(t, strings.Contains(startDownloadCommands, "/pulsar/download/java-function.jar"), "download command should contain /pulsar/download/java-function.jar: %s", startDownloadCommands)
148+
assert.Assert(t, strings.Contains(startFunctionCommands, "/pulsar/tmp/java-function.jar"), "function command should contain /pulsar/tmp/java-function.jar: %s", startFunctionCommands)
149+
}

‎controllers/spec/sink.go

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package spec
2020
import (
2121
"context"
2222
"regexp"
23+
"strings"
2324

2425
"github.com/streamnative/function-mesh/utils"
2526
"google.golang.org/protobuf/encoding/protojson"
@@ -106,7 +107,7 @@ func makeSinkContainer(sink *v1alpha1.Sink) *corev1.Container {
106107
allowPrivilegeEscalation := false
107108
mounts := makeSinkVolumeMounts(sink, sink.Spec.Pulsar.AuthConfig)
108109
if utils.EnableInitContainers {
109-
mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(sink.Spec.Java, nil, nil)...)
110+
mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(sink.Spec.Java, nil, nil, nil)...)
110111
}
111112
return &corev1.Container{
112113
// TODO new container to pull user code image and upload jars into bookkeeper
@@ -225,7 +226,17 @@ func MakeSinkCommand(sink *v1alpha1.Sink) []string {
225226
hasPulsarctl = true
226227
hasWget = true
227228
}
228-
return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
229+
mountPath := spec.Java.Jar
230+
if utils.EnableInitContainers {
231+
// for relative path, volume should be mounted to the WorkDir
232+
// and path also should be under the $WorkDir dir
233+
if !strings.HasPrefix(spec.Java.Jar, "/") {
234+
mountPath = WorkDir + spec.Java.Jar
235+
} else if !strings.HasPrefix(spec.Java.Jar, WorkDir) {
236+
mountPath = strings.Replace(spec.Java.Jar, "/", WorkDir, 1)
237+
}
238+
}
239+
return MakeJavaFunctionCommand(spec.Java.JarLocation, mountPath,
229240
spec.Name, spec.ClusterName,
230241
GenerateJavaLogConfigCommand(spec.Java, spec.LogTopicAgent),
231242
parseJavaLogLevel(spec.Java),

‎controllers/spec/source.go

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"context"
2222
"fmt"
2323
"regexp"
24+
"strings"
2425

2526
"github.com/streamnative/function-mesh/utils"
2627
"google.golang.org/protobuf/encoding/protojson"
@@ -102,7 +103,7 @@ func makeSourceContainer(source *v1alpha1.Source) *corev1.Container {
102103
allowPrivilegeEscalation := false
103104
mounts := makeSourceVolumeMounts(source, source.Spec.Pulsar.AuthConfig)
104105
if utils.EnableInitContainers {
105-
mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(source.Spec.Java, nil, nil)...)
106+
mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(source.Spec.Java, nil, nil, nil)...)
106107
}
107108
return &corev1.Container{
108109
// TODO new container to pull user code image and upload jars into bookkeeper
@@ -172,7 +173,18 @@ func makeSourceCommand(source *v1alpha1.Source) []string {
172173
hasPulsarctl = true
173174
hasWget = true
174175
}
175-
return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
176+
177+
mountPath := spec.Java.Jar
178+
if utils.EnableInitContainers {
179+
// for relative path, volume should be mounted to the WorkDir
180+
// and path also should be under the $WorkDir dir
181+
if !strings.HasPrefix(spec.Java.Jar, "/") {
182+
mountPath = WorkDir + spec.Java.Jar
183+
} else if !strings.HasPrefix(spec.Java.Jar, WorkDir) {
184+
mountPath = strings.Replace(spec.Java.Jar, "/", WorkDir, 1)
185+
}
186+
}
187+
return MakeJavaFunctionCommand(spec.Java.JarLocation, mountPath,
176188
spec.Name, spec.ClusterName,
177189
GenerateJavaLogConfigCommand(spec.Java, spec.LogTopicAgent),
178190
parseJavaLogLevel(spec.Java),

0 commit comments

Comments
 (0)