diff --git a/pkg/cmdutils/runtime_options.go b/pkg/cmdutils/runtime_options.go new file mode 100644 index 000000000..c55ddff73 --- /dev/null +++ b/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, + }) +} diff --git a/pkg/cmdutils/runtime_options_test.go b/pkg/cmdutils/runtime_options_test.go new file mode 100644 index 000000000..f969c9f50 --- /dev/null +++ b/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) +} diff --git a/pkg/cmdutils/verb.go b/pkg/cmdutils/verb.go index c5466e996..07baabded 100644 --- a/pkg/cmdutils/verb.go +++ b/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) diff --git a/pkg/ctl/functions/create.go b/pkg/ctl/functions/create.go index 23e49f1fb..bae8b8ab8 100644 --- a/pkg/ctl/functions/create.go +++ b/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() diff --git a/pkg/ctl/functions/functions.go b/pkg/ctl/functions/functions.go index e29fb3e95..bce3f22ec 100644 --- a/pkg/ctl/functions/functions.go +++ b/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) diff --git a/pkg/ctl/functions/runtime_options.go b/pkg/ctl/functions/runtime_options.go new file mode 100644 index 000000000..e16ed4d36 --- /dev/null +++ b/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 +} diff --git a/pkg/ctl/functions/runtime_options_test.go b/pkg/ctl/functions/runtime_options_test.go new file mode 100644 index 000000000..36d565104 --- /dev/null +++ b/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) +} diff --git a/pkg/ctl/functions/update.go b/pkg/ctl/functions/update.go index 9716b18b9..fe3dc3aa2 100644 --- a/pkg/ctl/functions/update.go +++ b/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() diff --git a/pkg/ctl/sinks/create.go b/pkg/ctl/sinks/create.go index 08bff7fb3..84bcd2974 100644 --- a/pkg/ctl/sinks/create.go +++ b/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() diff --git a/pkg/ctl/sinks/runtime_options.go b/pkg/ctl/sinks/runtime_options.go new file mode 100644 index 000000000..bafc309da --- /dev/null +++ b/pkg/ctl/sinks/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 sinks + +import ( + "fmt" + + util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils" + "github.com/streamnative/pulsarctl/pkg/cmdutils" +) + +func applySinkRuntimeOptions( + sinkData *util.SinkData, + runtimeOptions cmdutils.RuntimeOptionsConfig, + operation cmdutils.RuntimeOperation, +) error { + if runtimeOptions.CustomRuntimeOptionsInjector == nil || sinkData.SinkConf == nil { + return nil + } + + value, err := runtimeOptions.ApplyCustomRuntimeOptions( + cmdutils.RuntimeResourceSink, + operation, + sinkData.SinkConf.CustomRuntimeOptions, + ) + if err != nil { + return fmt.Errorf("inject sink %s custom runtime options: %w", operation, err) + } + sinkData.SinkConf.CustomRuntimeOptions = value + return nil +} diff --git a/pkg/ctl/sinks/runtime_options_test.go b/pkg/ctl/sinks/runtime_options_test.go new file mode 100644 index 000000000..13b2781e5 --- /dev/null +++ b/pkg/ctl/sinks/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 sinks + +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 TestApplySinkRuntimeOptionsUsesCurrentValue(t *testing.T) { + sinkData := &util.SinkData{SinkConf: &util.SinkConfig{CustomRuntimeOptions: `{"base":true}`}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(ctx cmdutils.CustomRuntimeOptionsContext) (string, error) { + assert.Equal(t, cmdutils.RuntimeResourceSink, ctx.Resource) + assert.Equal(t, cmdutils.RuntimeOperationCreate, ctx.Operation) + assert.Equal(t, `{"base":true}`, ctx.Current) + return `{"base":true,"extra":true}`, nil + }, + } + + err := applySinkRuntimeOptions(sinkData, runtimeOptions, cmdutils.RuntimeOperationCreate) + + assert.NoError(t, err) + assert.Equal(t, `{"base":true,"extra":true}`, sinkData.SinkConf.CustomRuntimeOptions) +} + +func TestApplySinkRuntimeOptionsCanClearValue(t *testing.T) { + sinkData := &util.SinkData{SinkConf: &util.SinkConfig{CustomRuntimeOptions: `{"base":true}`}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) { + return "", nil + }, + } + + err := applySinkRuntimeOptions(sinkData, runtimeOptions, cmdutils.RuntimeOperationUpdate) + + assert.NoError(t, err) + assert.Empty(t, sinkData.SinkConf.CustomRuntimeOptions) +} + +func TestApplySinkRuntimeOptionsWrapsInjectorError(t *testing.T) { + sinkData := &util.SinkData{SinkConf: &util.SinkConfig{CustomRuntimeOptions: "current"}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) { + return "", errors.New("boom") + }, + } + + err := applySinkRuntimeOptions(sinkData, runtimeOptions, cmdutils.RuntimeOperationUpdate) + + assert.EqualError(t, err, "inject sink update custom runtime options: boom") + assert.Equal(t, "current", sinkData.SinkConf.CustomRuntimeOptions) +} + +func TestApplySinkRuntimeOptionsNoInjectorLeavesValue(t *testing.T) { + sinkData := &util.SinkData{SinkConf: &util.SinkConfig{CustomRuntimeOptions: `{"base":true}`}} + + err := applySinkRuntimeOptions(sinkData, cmdutils.RuntimeOptionsConfig{}, cmdutils.RuntimeOperationCreate) + + assert.NoError(t, err) + assert.Equal(t, `{"base":true}`, sinkData.SinkConf.CustomRuntimeOptions) +} diff --git a/pkg/ctl/sinks/sinks.go b/pkg/ctl/sinks/sinks.go index fc18881c5..caa45342e 100644 --- a/pkg/ctl/sinks/sinks.go +++ b/pkg/ctl/sinks/sinks.go @@ -23,7 +23,8 @@ import ( "github.com/spf13/cobra" ) -func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { +func Command(flagGrouping *cmdutils.FlagGrouping, runtimeOptions ...cmdutils.RuntimeOptionsConfig) *cobra.Command { + resolvedRuntimeOptions := cmdutils.ResolveRuntimeOptions(runtimeOptions...) resourceCmd := cmdutils.NewResourceCmd( "sinks", "Interface for managing Pulsar IO sinks (egress data from Pulsar)", @@ -31,8 +32,8 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "sinks", ) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createSinksCmd) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateSinksCmd) + cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, createSinksCmd) + cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, updateSinksCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteSinksCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getSinksCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listSinksCmd) diff --git a/pkg/ctl/sinks/update.go b/pkg/ctl/sinks/update.go index bf5f64df9..d0e9a666d 100644 --- a/pkg/ctl/sinks/update.go +++ b/pkg/ctl/sinks/update.go @@ -337,6 +337,12 @@ func doUpdateSink(vc *cmdutils.VerbCmd, sinkData *util.SinkData) error { return err } + err = applySinkRuntimeOptions(sinkData, vc.RuntimeOptions, cmdutils.RuntimeOperationUpdate) + if err != nil { + _ = vc.Command.Help() + return err + } + checkArgsForUpdate(sinkData.SinkConf) // convert the map[interface{}]interface{} to a map[string]interface{} for unmarshal diff --git a/pkg/ctl/sources/create.go b/pkg/ctl/sources/create.go index 30d70329c..ccd5a7820 100644 --- a/pkg/ctl/sources/create.go +++ b/pkg/ctl/sources/create.go @@ -278,6 +278,12 @@ func doCreateSources(vc *cmdutils.VerbCmd, sourceData *util.SourceData) error { return err } + err = applySourceRuntimeOptions(sourceData, vc.RuntimeOptions, cmdutils.RuntimeOperationCreate) + if err != nil { + _ = vc.Command.Help() + return err + } + err = validateSourceConfigs(sourceData.SourceConf) if err != nil { _ = vc.Command.Help() diff --git a/pkg/ctl/sources/runtime_options.go b/pkg/ctl/sources/runtime_options.go new file mode 100644 index 000000000..d693b6b6d --- /dev/null +++ b/pkg/ctl/sources/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 sources + +import ( + "fmt" + + util "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils" + "github.com/streamnative/pulsarctl/pkg/cmdutils" +) + +func applySourceRuntimeOptions( + sourceData *util.SourceData, + runtimeOptions cmdutils.RuntimeOptionsConfig, + operation cmdutils.RuntimeOperation, +) error { + if runtimeOptions.CustomRuntimeOptionsInjector == nil || sourceData.SourceConf == nil { + return nil + } + + value, err := runtimeOptions.ApplyCustomRuntimeOptions( + cmdutils.RuntimeResourceSource, + operation, + sourceData.SourceConf.CustomRuntimeOptions, + ) + if err != nil { + return fmt.Errorf("inject source %s custom runtime options: %w", operation, err) + } + sourceData.SourceConf.CustomRuntimeOptions = value + return nil +} diff --git a/pkg/ctl/sources/runtime_options_test.go b/pkg/ctl/sources/runtime_options_test.go new file mode 100644 index 000000000..2314b7587 --- /dev/null +++ b/pkg/ctl/sources/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 sources + +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 TestApplySourceRuntimeOptionsUsesCurrentValue(t *testing.T) { + sourceData := &util.SourceData{SourceConf: &util.SourceConfig{CustomRuntimeOptions: `{"base":true}`}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(ctx cmdutils.CustomRuntimeOptionsContext) (string, error) { + assert.Equal(t, cmdutils.RuntimeResourceSource, ctx.Resource) + assert.Equal(t, cmdutils.RuntimeOperationCreate, ctx.Operation) + assert.Equal(t, `{"base":true}`, ctx.Current) + return `{"base":true,"extra":true}`, nil + }, + } + + err := applySourceRuntimeOptions(sourceData, runtimeOptions, cmdutils.RuntimeOperationCreate) + + assert.NoError(t, err) + assert.Equal(t, `{"base":true,"extra":true}`, sourceData.SourceConf.CustomRuntimeOptions) +} + +func TestApplySourceRuntimeOptionsCanClearValue(t *testing.T) { + sourceData := &util.SourceData{SourceConf: &util.SourceConfig{CustomRuntimeOptions: `{"base":true}`}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) { + return "", nil + }, + } + + err := applySourceRuntimeOptions(sourceData, runtimeOptions, cmdutils.RuntimeOperationUpdate) + + assert.NoError(t, err) + assert.Empty(t, sourceData.SourceConf.CustomRuntimeOptions) +} + +func TestApplySourceRuntimeOptionsWrapsInjectorError(t *testing.T) { + sourceData := &util.SourceData{SourceConf: &util.SourceConfig{CustomRuntimeOptions: "current"}} + runtimeOptions := cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: func(cmdutils.CustomRuntimeOptionsContext) (string, error) { + return "", errors.New("boom") + }, + } + + err := applySourceRuntimeOptions(sourceData, runtimeOptions, cmdutils.RuntimeOperationUpdate) + + assert.EqualError(t, err, "inject source update custom runtime options: boom") + assert.Equal(t, "current", sourceData.SourceConf.CustomRuntimeOptions) +} + +func TestApplySourceRuntimeOptionsNoInjectorLeavesValue(t *testing.T) { + sourceData := &util.SourceData{SourceConf: &util.SourceConfig{CustomRuntimeOptions: `{"base":true}`}} + + err := applySourceRuntimeOptions(sourceData, cmdutils.RuntimeOptionsConfig{}, cmdutils.RuntimeOperationCreate) + + assert.NoError(t, err) + assert.Equal(t, `{"base":true}`, sourceData.SourceConf.CustomRuntimeOptions) +} diff --git a/pkg/ctl/sources/sources.go b/pkg/ctl/sources/sources.go index db1210c33..3bcc8ff11 100644 --- a/pkg/ctl/sources/sources.go +++ b/pkg/ctl/sources/sources.go @@ -23,7 +23,8 @@ import ( "github.com/spf13/cobra" ) -func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { +func Command(flagGrouping *cmdutils.FlagGrouping, runtimeOptions ...cmdutils.RuntimeOptionsConfig) *cobra.Command { + resolvedRuntimeOptions := cmdutils.ResolveRuntimeOptions(runtimeOptions...) resourceCmd := cmdutils.NewResourceCmd( "sources", "Interface for managing Pulsar IO Sources (ingress data into Pulsar)", @@ -31,8 +32,8 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "sources", ) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createSourcesCmd) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateSourcesCmd) + cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, createSourcesCmd) + cmdutils.AddVerbCmdWithRuntimeOptions(flagGrouping, resourceCmd, resolvedRuntimeOptions, updateSourcesCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteSourcesCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getSourcesCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listSourcesCmd) diff --git a/pkg/ctl/sources/update.go b/pkg/ctl/sources/update.go index cfee4e528..a7960d819 100644 --- a/pkg/ctl/sources/update.go +++ b/pkg/ctl/sources/update.go @@ -257,6 +257,12 @@ func doUpdateSource(vc *cmdutils.VerbCmd, sourceData *util.SourceData) error { return err } + err = applySourceRuntimeOptions(sourceData, vc.RuntimeOptions, cmdutils.RuntimeOperationUpdate) + if err != nil { + _ = vc.Command.Help() + return err + } + checkArgsForUpdate(sourceData.SourceConf) // convert the map[interface{}]interface{} to a map[string]interface{} for unmarshal diff --git a/pkg/options.go b/pkg/options.go new file mode 100644 index 000000000..5b56eccc4 --- /dev/null +++ b/pkg/options.go @@ -0,0 +1,73 @@ +// 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 pkg + +import "github.com/streamnative/pulsarctl/pkg/cmdutils" + +// ResourceKind identifies a resource that supports custom runtime options injection. +type ResourceKind = cmdutils.RuntimeResource + +// RuntimeOperation identifies an operation that supports custom runtime options injection. +type RuntimeOperation = cmdutils.RuntimeOperation + +const ( + // ResourceFunction identifies Pulsar Functions commands. + ResourceFunction ResourceKind = cmdutils.RuntimeResourceFunction + // ResourceSink identifies Pulsar IO sink commands. + ResourceSink ResourceKind = cmdutils.RuntimeResourceSink + // ResourceSource identifies Pulsar IO source commands. + ResourceSource ResourceKind = cmdutils.RuntimeResourceSource + + // RuntimeOperationCreate identifies create operations. + RuntimeOperationCreate RuntimeOperation = cmdutils.RuntimeOperationCreate + // RuntimeOperationUpdate identifies update operations. + RuntimeOperationUpdate RuntimeOperation = cmdutils.RuntimeOperationUpdate +) + +// CustomRuntimeOptionsContext describes the current custom runtime options value +// before it is sent to the Pulsar admin API. +type CustomRuntimeOptionsContext = cmdutils.CustomRuntimeOptionsContext + +// CustomRuntimeOptionsInjector returns the final custom runtime options string +// for functions, sinks, and sources create/update commands. +type CustomRuntimeOptionsInjector = cmdutils.CustomRuntimeOptionsInjector + +// Option configures a pulsarctl root command. +type Option func(*options) + +type options struct { + customRuntimeOptionsInjector CustomRuntimeOptionsInjector +} + +func defaultOptions() options { + return options{} +} + +// WithCustomRuntimeOptionsInjector configures a hook that can replace the final +// customRuntimeOptions value for functions, sinks, and sources create/update commands. +func WithCustomRuntimeOptionsInjector(injector CustomRuntimeOptionsInjector) Option { + return func(opts *options) { + opts.customRuntimeOptionsInjector = injector + } +} + +func (o options) runtimeOptionsConfig() cmdutils.RuntimeOptionsConfig { + return cmdutils.RuntimeOptionsConfig{ + CustomRuntimeOptionsInjector: o.customRuntimeOptionsInjector, + } +} diff --git a/pkg/options_test.go b/pkg/options_test.go new file mode 100644 index 000000000..3f75d3b8d --- /dev/null +++ b/pkg/options_test.go @@ -0,0 +1,54 @@ +// 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 pkg + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNewPulsarctlCmdAcceptsCustomRuntimeOptionsInjector(t *testing.T) { + cmd := NewPulsarctlCmd(WithCustomRuntimeOptionsInjector( + func(ctx CustomRuntimeOptionsContext) (string, error) { + return ctx.Current, nil + }, + )) + + assert.NotNil(t, cmd) + assert.Equal(t, "pulsarctl [command]", cmd.Use) +} + +func TestWithCustomRuntimeOptionsInjectorSetsRuntimeOptionsConfig(t *testing.T) { + resolved := defaultOptions() + WithCustomRuntimeOptionsInjector(func(ctx CustomRuntimeOptionsContext) (string, error) { + assert.Equal(t, ResourceFunction, ctx.Resource) + assert.Equal(t, RuntimeOperationCreate, ctx.Operation) + assert.Equal(t, `{"base":true}`, ctx.Current) + return `{"base":true,"extra":true}`, nil + })(&resolved) + + value, err := resolved.runtimeOptionsConfig().ApplyCustomRuntimeOptions( + ResourceFunction, + RuntimeOperationCreate, + `{"base":true}`, + ) + + assert.NoError(t, err) + assert.Equal(t, `{"base":true,"extra":true}`, value) +} diff --git a/pkg/pulsarctl.go b/pkg/pulsarctl.go index 171bb20a7..c4f352b84 100644 --- a/pkg/pulsarctl.go +++ b/pkg/pulsarctl.go @@ -50,7 +50,15 @@ import ( lol "github.com/kris-nova/lolgopher" ) -func NewPulsarctlCmd() *cobra.Command { +func NewPulsarctlCmd(opts ...Option) *cobra.Command { + resolvedOptions := defaultOptions() + for _, opt := range opts { + if opt != nil { + opt(&resolvedOptions) + } + } + runtimeOptions := resolvedOptions.runtimeOptionsConfig() + var colorValue string flagGrouping := cmdutils.NewGrouping() @@ -115,9 +123,9 @@ func NewPulsarctlCmd() *cobra.Command { rootCmd.AddCommand(cluster.Command(flagGrouping)) rootCmd.AddCommand(tenant.Command(flagGrouping)) rootCmd.AddCommand(completion.Command(rootCmd)) - rootCmd.AddCommand(function.Command(flagGrouping)) - rootCmd.AddCommand(source.Command(flagGrouping)) - rootCmd.AddCommand(sink.Command(flagGrouping)) + rootCmd.AddCommand(function.Command(flagGrouping, runtimeOptions)) + rootCmd.AddCommand(source.Command(flagGrouping, runtimeOptions)) + rootCmd.AddCommand(sink.Command(flagGrouping, runtimeOptions)) rootCmd.AddCommand(topic.Command(flagGrouping)) rootCmd.AddCommand(namespace.Command(flagGrouping)) rootCmd.AddCommand(schema.Command(flagGrouping))