Add stats, status, querystate, putstate, trigger cmds for Pulsar Functions

.github/workflows/go.yml

@@ -19,7 +19,9 @@ jobs:
     - name: Get dependencies
       run: |
         docker pull apachepulsar/pulsar
-        docker run -d -p 8080:8080 -p 6650:6650 -v ${PWD}:/pulsar_test apachepulsar/pulsar bin/pulsar standalone
+        docker run -d -p 8080:8080 -p 6650:6650 -v ${PWD}:/pulsar_test --name pulsarctl apachepulsar/pulsar bin/pulsar standalone
+        docker cp pulsarctl:/pulsar/examples/api-examples.jar ${PWD}/test/functions/
+        ls -al test/functions/
         sleep 10
 
     - name: Build

.gitignore

@@ -5,6 +5,7 @@
 *.so
 *.dylib
 pulsarctl
+api-examples.jar
 
 # Test binary, build with `go test -c`
 *.test

pkg/ctl/functions/functions.go

@@ -39,6 +39,11 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statusFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, statsFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, querystateFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, putstateFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, triggerFunctionsCmd)
 
 	return resourceCmd
 }

pkg/ctl/functions/putstate.go

@@ -0,0 +1,142 @@
+// 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 (
+	"encoding/json"
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func putstateFunctionsCmd(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Put the state associated with a Pulsar Function."
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []pulsar.Example
+	putstate := pulsar.Example{
+		Desc: "Put the state associated with a Pulsar Function",
+		Command: "pulsarctl functions putstate \n" +
+			"\t--tenant public\n" +
+			"\t--namespace default\n" +
+			"\t--name <the name of Pulsar Function> \n" +
+			"\t--state \"{\"key\":\"pulsar\", \"stringValue\":\"hello\"}\" ",
+	}
+	examples = append(examples, putstate)
+
+	putstateWithFQFN := pulsar.Example{
+		Desc: "Put the state associated with a Pulsar Function with FQFN",
+		Command: "pulsarctl functions putstate \n" +
+			"\t--fqfn tenant/namespace/name [eg: public/default/ExampleFunctions] \n" +
+			"\t--state \"{\"key\":\"pulsar\", \"stringValue\":\"hello\"}\"",
+	}
+	examples = append(examples, putstateWithFQFN)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "PutState successfully",
+	}
+
+	failOut := pulsar.Output{
+		Desc: "You must specify a name for the Pulsar Functions or a FQFN, please check the --name args",
+		Out:  "[✖]  you must specify a name for the function or a Fully Qualified Function Name (FQFN)",
+	}
+
+	failOutWithNameNotExist := pulsar.Output{
+		Desc: "The name of Pulsar Functions doesn't exist, please check the --name args",
+		Out:  "[✖]  code: 404 reason: Function <your function name> doesn't exist",
+	}
+
+	out = append(out, successOut, failOut, failOutWithNameNotExist)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"putstate",
+		"Put the state associated with a Pulsar Function",
+		desc.ToString(),
+		"putstate",
+	)
+
+	functionData := &pulsar.FunctionData{}
+
+	// set the run function
+	vc.SetRunFunc(func() error {
+		return doPutStateFunction(vc, functionData)
+	})
+
+	// register the params
+	vc.FlagSetGroup.InFlagSet("FunctionsConfig", func(flagSet *pflag.FlagSet) {
+		flagSet.StringVar(
+			&functionData.FQFN,
+			"fqfn",
+			"",
+			"The Fully Qualified Function Name (FQFN) for the function")
+
+		flagSet.StringVar(
+			&functionData.Tenant,
+			"tenant",
+			"",
+			"The tenant of a Pulsar Function")
+
+		flagSet.StringVar(
+			&functionData.Namespace,
+			"namespace",
+			"",
+			"The namespace of a Pulsar Function")
+
+		flagSet.StringVar(
+			&functionData.FuncName,
+			"name",
+			"",
+			"The name of a Pulsar Function")
+
+		flagSet.StringVarP(
+			&functionData.State,
+			"state",
+			"t",
+			"",
+			"The FunctionState that needs to be put")
+	})
+}
+
+func doPutStateFunction(vc *cmdutils.VerbCmd, funcData *pulsar.FunctionData) error {
+	err := processBaseArguments(funcData)
+	if err != nil {
+		vc.Command.Help()
+		return err
+	}
+	admin := cmdutils.NewPulsarClientWithApiVersion(pulsar.V3)
+
+	var state pulsar.FunctionState
+	err = json.Unmarshal([]byte(funcData.State), &state)
+	if err != nil {
+		return err
+	}
+
+	err = admin.Functions().PutFunctionState(funcData.Tenant, funcData.Namespace, funcData.FuncName, state)
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		vc.Command.Printf("PutState successfully")
+	}
+
+	return err
+}

pkg/ctl/functions/putstate_test.go

@@ -0,0 +1,114 @@
+// 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 (
+	"encoding/json"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"github.com/stretchr/testify/assert"
+	"strings"
+	"testing"
+	`time`
+)
+
+func TestStateFunctions(t *testing.T) {
+	basePath, err := getDirHelp()
+	if basePath == "" || err != nil {
+		t.Error(err)
+	}
+	t.Logf("base path: %s", basePath)
+	args := []string{"create",
+		"--tenant", "public",
+		"--namespace", "default",
+		"--name", "test-functions-putstate",
+		"--inputs", "test-input-topic",
+		"--output", "persistent://public/default/test-output-topic",
+		"--classname", "org.apache.pulsar.functions.api.examples.ExclamationFunction",
+		"--jar", basePath + "/test/functions/api-examples.jar",
+	}
+
+	out, execErr, err := TestFunctionsCommands(createFunctionsCmd, args)
+	assert.Nil(t, err)
+	if execErr != nil {
+		t.Errorf("create functions error value: %s", execErr.Error())
+	}
+	assert.Equal(t, out.String(), "Created test-functions-putstate successfully")
+
+	// wait the function create successfully
+	time.Sleep(time.Second * 20)
+
+	putstateArgs := []string{"putstate",
+		"--tenant", "public",
+		"--namespace", "default",
+		"--name", "test-functions-putstate",
+		"--state", "{\"key\":\"pulsar\", \"stringValue\":\"hello\"}",
+	}
+
+	outPutState, execE, err := TestFunctionsCommands(putstateFunctionsCmd, putstateArgs)
+	assert.Nil(t, err)
+	if execE != nil {
+		t.Errorf("put state functions error value: %s ", execE.Error())
+	}
+	t.Logf("outPutState:%s", outPutState.String())
+	assert.Equal(t, outPutState.String(), "PutState successfully")
+
+	// test failure case for put state
+	failureStateArgs := []string{"putstate",
+		"--name", "not-exist",
+		"--state", "{\"key\":\"pulsar\", \"stringValue\":\"hello\"}",
+	}
+
+	_, execErrMsg, _ := TestFunctionsCommands(putstateFunctionsCmd, failureStateArgs)
+	assert.NotNil(t, execErrMsg)
+	exceptMsg := "'not-exist' is not found"
+	assert.True(t, strings.Contains(execErrMsg.Error(), exceptMsg))
+
+	// query state
+	queryStateArgs := []string{"querystate",
+		"--tenant", "public",
+		"--namespace", "default",
+		"--name", "test-functions-putstate",
+		"--key", "pulsar",
+	}
+
+	outQueryState, _, err := TestFunctionsCommands(querystateFunctionsCmd, queryStateArgs)
+	assert.Nil(t, err)
+	t.Logf("outQueryState:%s", outQueryState.String())
+
+	var state pulsar.FunctionState
+	err = json.Unmarshal(outQueryState.Bytes(), &state)
+	assert.Nil(t, err)
+
+	assert.Equal(t, "pulsar", state.Key)
+	assert.Equal(t, "hello", state.StringValue)
+	assert.Equal(t, int64(0), state.Version)
+	// put state again
+	outPutStateAgain, _, err := TestFunctionsCommands(putstateFunctionsCmd, putstateArgs)
+	assert.Nil(t, err)
+	assert.Equal(t, outPutStateAgain.String(), "PutState successfully")
+
+	// query state again
+	outQueryStateAgain, _, err := TestFunctionsCommands(querystateFunctionsCmd, queryStateArgs)
+	assert.Nil(t, err)
+
+	var stateAgain pulsar.FunctionState
+	err = json.Unmarshal(outQueryStateAgain.Bytes(), &stateAgain)
+	assert.Nil(t, err)
+
+	assert.Equal(t, int64(1), stateAgain.Version)
+}