feat(cmd): add custom runtime options injection

pkg/cmdutils/runtime_options.go

@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 cmdutils
+
+// RuntimeResource identifies a runtime-capable command resource.
+type RuntimeResource string
+
+// RuntimeOperation identifies a command operation that can apply runtime options.
+type RuntimeOperation string
+
+const (
+	// RuntimeResourceFunction identifies Pulsar Functions commands.
+	RuntimeResourceFunction RuntimeResource = "function"
+	// RuntimeResourceSink identifies Pulsar IO sink commands.
+	RuntimeResourceSink RuntimeResource = "sink"
+	// RuntimeResourceSource identifies Pulsar IO source commands.
+	RuntimeResourceSource RuntimeResource = "source"
+
+	// RuntimeOperationCreate identifies create operations.
+	RuntimeOperationCreate RuntimeOperation = "create"
+	// RuntimeOperationUpdate identifies update operations.
+	RuntimeOperationUpdate RuntimeOperation = "update"
+)
+
+// CustomRuntimeOptionsContext describes the current custom runtime options value
+// before it is sent to the Pulsar admin API.
+type CustomRuntimeOptionsContext struct {
+	Resource  RuntimeResource
+	Operation RuntimeOperation
+	Current   string
+}
+
+// CustomRuntimeOptionsInjector returns the final custom runtime options string
+// for a runtime-capable resource command.
+type CustomRuntimeOptionsInjector func(CustomRuntimeOptionsContext) (string, error)
+
+// RuntimeOptionsConfig contains immutable runtime option hooks for a command tree.
+type RuntimeOptionsConfig struct {
+	CustomRuntimeOptionsInjector CustomRuntimeOptionsInjector
+}
+
+// ResolveRuntimeOptions merges optional runtime option configs.
+func ResolveRuntimeOptions(runtimeOptions ...RuntimeOptionsConfig) RuntimeOptionsConfig {
+	var resolved RuntimeOptionsConfig
+	for _, opts := range runtimeOptions {
+		if opts.CustomRuntimeOptionsInjector != nil {
+			resolved.CustomRuntimeOptionsInjector = opts.CustomRuntimeOptionsInjector
+		}
+	}
+	return resolved
+}
+
+// ApplyCustomRuntimeOptions applies the configured custom runtime options injector.
+func (c RuntimeOptionsConfig) ApplyCustomRuntimeOptions(
+	resource RuntimeResource,
+	operation RuntimeOperation,
+	current string,
+) (string, error) {
+	if c.CustomRuntimeOptionsInjector == nil {
+		return current, nil
+	}
+	return c.CustomRuntimeOptionsInjector(CustomRuntimeOptionsContext{
+		Resource:  resource,
+		Operation: operation,
+		Current:   current,
+	})
+}

pkg/cmdutils/runtime_options_test.go

@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 cmdutils
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRuntimeOptionsConfigApplyCustomRuntimeOptionsNoInjector(t *testing.T) {
+	value, err := (RuntimeOptionsConfig{}).ApplyCustomRuntimeOptions(
+		RuntimeResourceFunction,
+		RuntimeOperationCreate,
+		`{"base":true}`,
+	)
+
+	assert.NoError(t, err)
+	assert.Equal(t, `{"base":true}`, value)
+}
+
+func TestResolveRuntimeOptionsUsesLastInjector(t *testing.T) {
+	first := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
+		return "first", nil
+	}}
+	second := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
+		return "second", nil
+	}}
+
+	value, err := ResolveRuntimeOptions(first, second).ApplyCustomRuntimeOptions(
+		RuntimeResourceSource,
+		RuntimeOperationUpdate,
+		"current",
+	)
+
+	assert.NoError(t, err)
+	assert.Equal(t, "second", value)
+}
+
+func TestAddVerbCmdWithRuntimeOptionsAssignsVerbConfig(t *testing.T) {
+	flagGrouping := NewGrouping()
+	parent := NewResourceCmd("resource", "short", "long")
+	captured := RuntimeOptionsConfig{}
+	expected := RuntimeOptionsConfig{CustomRuntimeOptionsInjector: func(CustomRuntimeOptionsContext) (string, error) {
+		return "injected", nil
+	}}
+
+	AddVerbCmdWithRuntimeOptions(flagGrouping, parent, expected, func(vc *VerbCmd) {
+		captured = vc.RuntimeOptions
+		vc.SetDescription("create", "create", "create", "")
+	})
+
+	value, err := captured.ApplyCustomRuntimeOptions(RuntimeResourceSink, RuntimeOperationCreate, "current")
+	assert.NoError(t, err)
+	assert.Equal(t, "injected", value)
+	assert.Len(t, parent.Commands(), 1)
+}

pkg/cmdutils/verb.go

@@ -34,12 +34,24 @@ type VerbCmd struct {
 	NameError             error // for testing
 	OutputConfig          *OutputConfig
 	ClusterConfigOverride *ClusterConfig
+	RuntimeOptions        RuntimeOptionsConfig
 }
 
 // AddVerbCmd create a registers a new command under the given resource command
 func AddVerbCmd(flagGrouping *FlagGrouping, parentResourceCmd *cobra.Command, newVerbCmd func(*VerbCmd)) {
+	AddVerbCmdWithRuntimeOptions(flagGrouping, parentResourceCmd, RuntimeOptionsConfig{}, newVerbCmd)
+}
+
+// AddVerbCmdWithRuntimeOptions creates and registers a new command under the given resource command.
+func AddVerbCmdWithRuntimeOptions(
+	flagGrouping *FlagGrouping,
+	parentResourceCmd *cobra.Command,
+	runtimeOptions RuntimeOptionsConfig,
+	newVerbCmd func(*VerbCmd),
+) {
 	verb := &VerbCmd{
-		Command: &cobra.Command{},
+		Command:        &cobra.Command{},
+		RuntimeOptions: ResolveRuntimeOptions(runtimeOptions),
 	}
 	verb.FlagSetGroup = flagGrouping.New(verb.Command)
 	newVerbCmd(verb)

pkg/ctl/functions/create.go

@@ -502,6 +502,12 @@ func doCreateFunctions(vc *cmdutils.VerbCmd, funcData *util.FunctionData) error
 		return err
 	}
 
+	err = applyFunctionRuntimeOptions(funcData, vc.RuntimeOptions, cmdutils.RuntimeOperationCreate)
+	if err != nil {
+		_ = vc.Command.Help()
+		return err
+	}
+
 	err = validateFunctionConfigs(funcData.FuncConf)
 	if err != nil {
 		_ = vc.Command.Help()

pkg/ctl/functions/functions.go

@@ -32,7 +32,8 @@ var checkPutStateArgs = func(args []string) error {
 	return nil
 }
 
-func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
+func Command(flagGrouping *cmdutils.FlagGrouping, runtimeOptions ...cmdutils.RuntimeOptionsConfig) *cobra.Command {
+	resolvedRuntimeOptions := cmdutils.ResolveRuntimeOptions(runtimeOptions...)
 	resourceCmd := cmdutils.NewResourceCmd(
 		"functions",
 		"Interface for managing Pulsar Functions "+
@@ -41,14 +42,14 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 		"pf",
 	)
 
-	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createFunctionsCmd)
+	cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, createFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, stopFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, startFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, restartFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getFunctionsCmd)
-	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateFunctionsCmd)
+	cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, updateFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statusFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statsFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, querystateFunctionsCmd)

pkg/ctl/functions/runtime_options.go

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 functions
+
+import (
+	"fmt"
+
+	util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+)
+
+func applyFunctionRuntimeOptions(
+	funcData *util.FunctionData,
+	runtimeOptions cmdutils.RuntimeOptionsConfig,
+	operation cmdutils.RuntimeOperation,
+) error {
+	if runtimeOptions.CustomRuntimeOptionsInjector == nil || funcData.FuncConf == nil {
+		return nil
+	}
+
+	value, err := runtimeOptions.ApplyCustomRuntimeOptions(
+		cmdutils.RuntimeResourceFunction,
+		operation,
+		funcData.FuncConf.CustomRuntimeOptions,
+	)
+	if err != nil {
+		return fmt.Errorf("inject function %s custom runtime options: %w", operation, err)
+	}
+	funcData.FuncConf.CustomRuntimeOptions = value
+	return nil
+}

pkg/ctl/functions/runtime_options_test.go

@@ -0,0 +1,81 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 functions
+
+import (
+	"errors"
+	"testing"
+
+	util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestApplyFunctionRuntimeOptionsUsesCurrentValue(t *testing.T) {
+	funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
+	runtimeOptions := cmdutils.RuntimeOptionsConfig{
+		CustomRuntimeOptionsInjector: func(ctx cmdutils.CustomRuntimeOptionsContext) (string, error) {
+			assert.Equal(t, cmdutils.RuntimeResourceFunction, ctx.Resource)
+			assert.Equal(t, cmdutils.RuntimeOperationCreate, ctx.Operation)
+			assert.Equal(t, `{"base":true}`, ctx.Current)
+			return `{"base":true,"extra":true}`, nil
+		},
+	}
+
+	err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationCreate)
+
+	assert.NoError(t, err)
+	assert.Equal(t, `{"base":true,"extra":true}`, funcData.FuncConf.CustomRuntimeOptions)
+}
+
+func TestApplyFunctionRuntimeOptionsCanClearValue(t *testing.T) {
+	funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
+	runtimeOptions := cmdutils.RuntimeOptionsConfig{
+		CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) {
+			return "", nil
+		},
+	}
+
+	err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationUpdate)
+
+	assert.NoError(t, err)
+	assert.Empty(t, funcData.FuncConf.CustomRuntimeOptions)
+}
+
+func TestApplyFunctionRuntimeOptionsWrapsInjectorError(t *testing.T) {
+	funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: "current"}}
+	runtimeOptions := cmdutils.RuntimeOptionsConfig{
+		CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) {
+			return "", errors.New("boom")
+		},
+	}
+
+	err := applyFunctionRuntimeOptions(funcData, runtimeOptions, cmdutils.RuntimeOperationUpdate)
+
+	assert.EqualError(t, err, "inject function update custom runtime options: boom")
+	assert.Equal(t, "current", funcData.FuncConf.CustomRuntimeOptions)
+}
+
+func TestApplyFunctionRuntimeOptionsNoInjectorLeavesValue(t *testing.T) {
+	funcData := &util.FunctionData{FuncConf: &util.FunctionConfig{CustomRuntimeOptions: `{"base":true}`}}
+
+	err := applyFunctionRuntimeOptions(funcData, cmdutils.RuntimeOptionsConfig{}, cmdutils.RuntimeOperationCreate)
+
+	assert.NoError(t, err)
+	assert.Equal(t, `{"base":true}`, funcData.FuncConf.CustomRuntimeOptions)
+}

pkg/ctl/functions/update.go

@@ -462,6 +462,12 @@ func doUpdateFunctions(vc *cmdutils.VerbCmd, funcData *util.FunctionData) error
 		return err
 	}
 
+	err = applyFunctionRuntimeOptions(funcData, vc.RuntimeOptions, cmdutils.RuntimeOperationUpdate)
+	if err != nil {
+		_ = vc.Command.Help()
+		return err
+	}
+
 	err = checkArgsForUpdate(funcData.FuncConf)
 	if err != nil {
 		_ = vc.Command.Help()

pkg/ctl/sinks/create.go

@@ -349,6 +349,12 @@ func doCreateSinks(vc *cmdutils.VerbCmd, sinkData *util.SinkData) error {
 		return err
 	}
 
+	err = applySinkRuntimeOptions(sinkData, vc.RuntimeOptions, cmdutils.RuntimeOperationCreate)
+	if err != nil {
+		_ = vc.Command.Help()
+		return err
+	}
+
 	err = validateSinkConfigs(sinkData.SinkConf)
 	if err != nil {
 		_ = vc.Command.Help()