Skip to content

Commit da10b8f

Browse files
freeznetfreeznet
authored
feat(cmd): add custom runtime options injection (#2103)
1 parent 798c19a commit da10b8f

21 files changed

Lines changed: 735 additions & 14 deletions

‎pkg/cmdutils/runtime_options.go‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package cmdutils
19+
20+
// RuntimeResource identifies a runtime-capable command resource.
21+
type RuntimeResource string
22+
23+
// RuntimeOperation identifies a command operation that can apply runtime options.
24+
type RuntimeOperation string
25+
26+
const (
27+
// RuntimeResourceFunction identifies Pulsar Functions commands.
28+
RuntimeResourceFunction RuntimeResource = "function"
29+
// RuntimeResourceSink identifies Pulsar IO sink commands.
30+
RuntimeResourceSink RuntimeResource = "sink"
31+
// RuntimeResourceSource identifies Pulsar IO source commands.
32+
RuntimeResourceSource RuntimeResource = "source"
33+
34+
// RuntimeOperationCreate identifies create operations.
35+
RuntimeOperationCreate RuntimeOperation = "create"
36+
// RuntimeOperationUpdate identifies update operations.
37+
RuntimeOperationUpdate RuntimeOperation = "update"
38+
)
39+
40+
// CustomRuntimeOptionsContext describes the current custom runtime options value
41+
// before it is sent to the Pulsar admin API.
42+
type CustomRuntimeOptionsContext struct {
43+
Resource RuntimeResource
44+
Operation RuntimeOperation
45+
Current string
46+
}
47+
48+
// CustomRuntimeOptionsInjector returns the final custom runtime options string
49+
// for a runtime-capable resource command.
50+
type CustomRuntimeOptionsInjector func(CustomRuntimeOptionsContext) (string, error)
51+
52+
// RuntimeOptionsConfig contains immutable runtime option hooks for a command tree.
53+
type RuntimeOptionsConfig struct {
54+
CustomRuntimeOptionsInjector CustomRuntimeOptionsInjector
55+
}
56+
57+
// ResolveRuntimeOptions merges optional runtime option configs.
58+
func ResolveRuntimeOptions(runtimeOptions ...RuntimeOptionsConfig) RuntimeOptionsConfig {
59+
var resolved RuntimeOptionsConfig
60+
for _, opts := range runtimeOptions {
61+
if opts.CustomRuntimeOptionsInjector != nil {
62+
resolved.CustomRuntimeOptionsInjector = opts.CustomRuntimeOptionsInjector
63+
}
64+
}
65+
return resolved
66+
}
67+
68+
// ApplyCustomRuntimeOptions applies the configured custom runtime options injector.
69+
func (c RuntimeOptionsConfig) ApplyCustomRuntimeOptions(
70+
resource RuntimeResource,
71+
operation RuntimeOperation,
72+
current string,
73+
) (string, error) {
74+
if c.CustomRuntimeOptionsInjector == nil {
75+
return current, nil
76+
}
77+
return c.CustomRuntimeOptionsInjector(CustomRuntimeOptionsContext{
78+
Resource: resource,
79+
Operation: operation,
80+
Current: current,
81+
})
82+
}

‎pkg/cmdutils/runtime_options_test.go‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package cmdutils
19+
20+
import (
21+
"testing"
22+
23+
"github.com/stretchr/testify/assert"
24+
)
25+
26+
func TestRuntimeOptionsConfigApplyCustomRuntimeOptionsNoInjector(t *testing.T) {
27+
value, err := (RuntimeOptionsConfig{}).ApplyCustomRuntimeOptions(
28+
RuntimeResourceFunction,
29+
RuntimeOperationCreate,
30+
`{"base":true}`,
31+
)
32+
33+
assert.NoError(t, err)
34+
assert.Equal(t, `{"base":true}`, value)
35+
}
36+
37+
func TestResolveRuntimeOptionsUsesLastInjector(t *testing.T) {
38+
first := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
39+
return "first", nil
40+
}}
41+
second := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
42+
return "second", nil
43+
}}
44+
45+
value, err := ResolveRuntimeOptions(first, second).ApplyCustomRuntimeOptions(
46+
RuntimeResourceSource,
47+
RuntimeOperationUpdate,
48+
"current",
49+
)
50+
51+
assert.NoError(t, err)
52+
assert.Equal(t, "second", value)
53+
}
54+
55+
func TestAddVerbCmdWithRuntimeOptionsAssignsVerbConfig(t *testing.T) {
56+
flagGrouping := NewGrouping()
57+
parent := NewResourceCmd("resource", "short", "long")
58+
captured := RuntimeOptionsConfig{}
59+
expected := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
60+
return "injected", nil
61+
}}
62+
63+
AddVerbCmdWithRuntimeOptions(flagGrouping, parent, expected, func(vc *VerbCmd) {
64+
captured = vc.RuntimeOptions
65+
vc.SetDescription("create", "create", "create", "")
66+
})
67+
68+
value, err := captured.ApplyCustomRuntimeOptions(RuntimeResourceSink, RuntimeOperationCreate, "current")
69+
assert.NoError(t, err)
70+
assert.Equal(t, "injected", value)
71+
assert.Len(t, parent.Commands(), 1)
72+
}

‎pkg/cmdutils/verb.go‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,24 @@ type VerbCmd struct {
3434
NameError error // for testing
3535
OutputConfig *OutputConfig
3636
ClusterConfigOverride *ClusterConfig
37+
RuntimeOptions RuntimeOptionsConfig
3738
}
3839

3940
// AddVerbCmd create a registers a new command under the given resource command
4041
func AddVerbCmd(flagGrouping *FlagGrouping, parentResourceCmd *cobra.Command, newVerbCmd func(*VerbCmd)) {
42+
AddVerbCmdWithRuntimeOptions(flagGrouping, parentResourceCmd, RuntimeOptionsConfig{}, newVerbCmd)
43+
}
44+
45+
// AddVerbCmdWithRuntimeOptions creates and registers a new command under the given resource command.
46+
func AddVerbCmdWithRuntimeOptions(
47+
flagGrouping *FlagGrouping,
48+
parentResourceCmd *cobra.Command,
49+
runtimeOptions RuntimeOptionsConfig,
50+
newVerbCmd func(*VerbCmd),
51+
) {
4152
verb := &VerbCmd{
42-
Command: &cobra.Command{},
53+
Command: &cobra.Command{},
54+
RuntimeOptions: ResolveRuntimeOptions(runtimeOptions),
4355
}
4456
verb.FlagSetGroup = flagGrouping.New(verb.Command)
4557
newVerbCmd(verb)

‎pkg/ctl/functions/create.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -502,6 +502,12 @@ func doCreateFunctions(vc *cmdutils.VerbCmd, funcData *util.FunctionData) error
502502
return err
503503
}
504504

505+
err = applyFunctionRuntimeOptions(funcData, vc.RuntimeOptions, cmdutils.RuntimeOperationCreate)
506+
if err != nil {
507+
_ = vc.Command.Help()
508+
return err
509+
}
510+
505511
err = validateFunctionConfigs(funcData.FuncConf)
506512
if err != nil {
507513
_ = vc.Command.Help()

‎pkg/ctl/functions/functions.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@ var checkPutStateArgs = func(args []string) error {
3232
return nil
3333
}
3434

35-
func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
35+
func Command(flagGrouping *cmdutils.FlagGrouping, runtimeOptions ...cmdutils.RuntimeOptionsConfig) *cobra.Command {
36+
resolvedRuntimeOptions := cmdutils.ResolveRuntimeOptions(runtimeOptions...)
3637
resourceCmd := cmdutils.NewResourceCmd(
3738
"functions",
3839
"Interface for managing Pulsar Functions "+
@@ -41,14 +42,14 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
4142
"pf",
4243
)
4344

44-
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createFunctionsCmd)
45+
cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, createFunctionsCmd)
4546
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, stopFunctionsCmd)
4647
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteFunctionsCmd)
4748
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, startFunctionsCmd)
4849
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, restartFunctionsCmd)
4950
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listFunctionsCmd)
5051
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getFunctionsCmd)
51-
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateFunctionsCmd)
52+
cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, updateFunctionsCmd)
5253
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statusFunctionsCmd)
5354
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statsFunctionsCmd)
5455
cmdutils.AddVerbCmd(flagGrouping, resourceCmd, querystateFunctionsCmd)

‎pkg/ctl/functions/runtime_options.go‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package functions
19+
20+
import (
21+
"fmt"
22+
23+
util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
24+
"github.com/streamnative/pulsarctl/pkg/cmdutils"
25+
)
26+
27+
func applyFunctionRuntimeOptions(
28+
funcData *util.FunctionData,
29+
runtimeOptions cmdutils.RuntimeOptionsConfig,
30+
operation cmdutils.RuntimeOperation,
31+
) error {
32+
if runtimeOptions.CustomRuntimeOptionsInjector == nil || funcData.FuncConf == nil {
33+
return nil
34+
}
35+
36+
value, err := runtimeOptions.ApplyCustomRuntimeOptions(
37+
cmdutils.RuntimeResourceFunction,
38+
operation,
39+
funcData.FuncConf.CustomRuntimeOptions,
40+
)
41+
if err != nil {
42+
return fmt.Errorf("inject function %s custom runtime options: %w", operation, err)
43+
}
44+
funcData.FuncConf.CustomRuntimeOptions = value
45+
return nil
46+
}

‎pkg/ctl/functions/runtime_options_test.go‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package functions
19+
20+
import (
21+
"errors"
22+
"testing"
23+
24+
util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
25+
"github.com/streamnative/pulsarctl/pkg/cmdutils"
26+
"github.com/stretchr/testify/assert"
27+
)
28+
29+
func TestApplyFunctionRuntimeOptionsUsesCurrentValue(t *testing.T) {
30+
funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
31+
runtimeOptions := cmdutils.RuntimeOptionsConfig{
32+
CustomRuntimeOptionsInjector: func(ctx cmdutils.CustomRuntimeOptionsContext) (string, error) {
33+
assert.Equal(t, cmdutils.RuntimeResourceFunction, ctx.Resource)
34+
assert.Equal(t, cmdutils.RuntimeOperationCreate, ctx.Operation)
35+
assert.Equal(t, `{"base":true}`, ctx.Current)
36+
return `{"base":true,"extra":true}`, nil
37+
},
38+
}
39+
40+
err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationCreate)
41+
42+
assert.NoError(t, err)
43+
assert.Equal(t, `{"base":true,"extra":true}`, funcData.FuncConf.CustomRuntimeOptions)
44+
}
45+
46+
func TestApplyFunctionRuntimeOptionsCanClearValue(t *testing.T) {
47+
funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
48+
runtimeOptions := cmdutils.RuntimeOptionsConfig{
49+
CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) {
50+
return "", nil
51+
},
52+
}
53+
54+
err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationUpdate)
55+
56+
assert.NoError(t, err)
57+
assert.Empty(t, funcData.FuncConf.CustomRuntimeOptions)
58+
}
59+
60+
func TestApplyFunctionRuntimeOptionsWrapsInjectorError(t *testing.T) {
61+
funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: "current"}}
62+
runtimeOptions := cmdutils.RuntimeOptionsConfig{
63+
CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) {
64+
return "", errors.New("boom")
65+
},
66+
}
67+
68+
err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationUpdate)
69+
70+
assert.EqualError(t, err, "inject function update custom runtime options: boom")
71+
assert.Equal(t, "current", funcData.FuncConf.CustomRuntimeOptions)
72+
}
73+
74+
func TestApplyFunctionRuntimeOptionsNoInjectorLeavesValue(t *testing.T) {
75+
funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
76+
77+
err := applyFunctionRuntimeOptions(funcData, cmdutils.RuntimeOptionsConfig{}, cmdutils.RuntimeOperationCreate)
78+
79+
assert.NoError(t, err)
80+
assert.Equal(t, `{"base":true}`, funcData.FuncConf.CustomRuntimeOptions)
81+
}

‎pkg/ctl/functions/update.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -462,6 +462,12 @@ func doUpdateFunctions(vc *cmdutils.VerbCmd, funcData *util.FunctionData) error
462462
return err
463463
}
464464

465+
err = applyFunctionRuntimeOptions(funcData, vc.RuntimeOptions, cmdutils.RuntimeOperationUpdate)
466+
if err != nil {
467+
_ = vc.Command.Help()
468+
return err
469+
}
470+
465471
err = checkArgsForUpdate(funcData.FuncConf)
466472
if err != nil {
467473
_ = vc.Command.Help()

‎pkg/ctl/sinks/create.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -349,6 +349,12 @@ func doCreateSinks(vc *cmdutils.VerbCmd, sinkData *util.SinkData) error {
349349
return err
350350
}
351351

352+
err = applySinkRuntimeOptions(sinkData, vc.RuntimeOptions, cmdutils.RuntimeOperationCreate)
353+
if err != nil {
354+
_ = vc.Command.Help()
355+
return err
356+
}
357+
352358
err = validateSinkConfigs(sinkData.SinkConf)
353359
if err != nil {
354360
_ = vc.Command.Help()

0 commit comments

Comments
 (0)