Add download command for functions CLI (#221)

Signed-off-by: xiaolong.ran <[email protected]>

Author: wolfstudy

go.mod

@@ -3,6 +3,7 @@ module github.com/streamnative/pulsarctl
 go 1.12
 
 require (
+	github.com/davecgh/go-spew v1.1.1
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/docker/go-connections v0.4.0
 	github.com/fatih/color v1.7.0

pkg/ctl/functions/download.go

@@ -0,0 +1,141 @@
+// 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 (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar/common"
+	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
+
+	"github.com/spf13/pflag"
+)
+
+func downloadFunctionsCmd(vc *cmdutils.VerbCmd) {
+	desc := cmdutils.LongDescription{}
+	desc.CommandUsedFor = "This command is used for download File Data from Pulsar."
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []cmdutils.Example
+	download := cmdutils.Example{
+		Desc: "Download File Data from Pulsar",
+		Command: "pulsarctl functions download \n" +
+			"\t--destination-file public\n" +
+			"\t--path default\n",
+	}
+	downloadByNs := cmdutils.Example{
+		Desc: "Download File Data from Pulsar",
+		Command: "pulsarctl functions download \n" +
+			"\t--destination-file public\n" +
+			"\t--tenant public\n" +
+			"\t--namespace default\n" +
+			"\t--name <function-name>\n",
+	}
+	examples = append(examples, download, downloadByNs)
+	desc.CommandExamples = examples
+
+	var out []cmdutils.Output
+	successOut := cmdutils.Output{
+		Desc: "normal output",
+		Out:  "Downloaded <the name of a Pulsar Function> successfully",
+	}
+	failOut := cmdutils.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)",
+	}
+
+	out = append(out, successOut, failOut)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"download",
+		"Download File Data from Pulsar",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"download",
+	)
+
+	functionData := &utils.FunctionData{}
+
+	// set the run function
+	vc.SetRunFunc(func() error {
+		return doDownloadFunctions(vc, functionData)
+	})
+
+	// register the params
+	vc.FlagSetGroup.InFlagSet("FunctionsConfig", func(flagSet *pflag.FlagSet) {
+		flagSet.StringVar(
+			&functionData.DestinationFile,
+			"destination-file",
+			"",
+			"The file to store downloaded content")
+
+		flagSet.StringVar(
+			&functionData.Path,
+			"path",
+			"",
+			"Path to store the content")
+
+		flagSet.StringVar(
+			&functionData.FQFN,
+			"fqfn",
+			"",
+			"The Fully Qualified Function Name (FQFN) for the function")
+
+		flagSet.StringVar(
+			&functionData.Tenant,
+			"tenant",
+			"",
+			"Tenant name")
+
+		flagSet.StringVar(
+			&functionData.Namespace,
+			"namespace",
+			"",
+			"Namespace name")
+
+		flagSet.StringVar(
+			&functionData.FuncName,
+			"name",
+			"",
+			"Function name")
+	})
+}
+
+func doDownloadFunctions(vc *cmdutils.VerbCmd, funcData *utils.FunctionData) error {
+	err := processBaseArguments(funcData)
+	if err != nil {
+		return err
+	}
+	admin := cmdutils.NewPulsarClientWithAPIVersion(common.V3)
+
+	if funcData.Path != "" {
+		err := admin.Functions().DownloadFunction(funcData.DestinationFile, funcData.Path)
+		if err != nil {
+			return err
+		}
+	} else {
+		err := admin.Functions().DownloadFunctionByNs(funcData.DestinationFile, funcData.Tenant,
+			funcData.Namespace, funcData.FuncName)
+		if err != nil {
+			return err
+		}
+	}
+
+	vc.Command.Printf("Downloaded %s successfully\n", funcData.DestinationFile)
+	return nil
+}

pkg/ctl/functions/download_test.go

@@ -0,0 +1,69 @@
+// 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 (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDownloadFunctions(t *testing.T) {
+	jarName := "dummyExample.jar"
+	_, err := os.Create(jarName)
+	assert.Nil(t, err)
+
+	defer os.Remove(jarName)
+
+	args := []string{"create",
+		"--tenant", "public",
+		"--namespace", "default",
+		"--name", "test-functions-download",
+		"--inputs", "test-input-topic",
+		"--output", "persistent://public/default/test-output-topic",
+		"--classname", "org.apache.pulsar.functions.api.examples.ExclamationFunction",
+		"--jar", jarName,
+	}
+
+	_, _, err = TestFunctionsCommands(createFunctionsCmd, args)
+	assert.Nil(t, err)
+
+	downloadArgs := []string{"download",
+		"--tenant", "public",
+		"--namespace", "default",
+		"--name", "test-functions-download",
+		"--destination-file", "./dummyExample.jar",
+	}
+	outPut, execErr, err := TestFunctionsCommands(downloadFunctionsCmd, downloadArgs)
+	assert.Equal(t, outPut.String(), "Downloaded ./dummyExample.jar successfully\n")
+	assert.Nil(t, execErr)
+	assert.Nil(t, err)
+}
+
+func TestDownloadFunctionsWithFailure(t *testing.T) {
+	notExistNameOrFqfnArgs := []string{"download",
+		"--tenant", "public",
+		"--namespace", "default",
+	}
+	_, execErrMsg, _ := TestFunctionsCommands(downloadFunctionsCmd, notExistNameOrFqfnArgs)
+	failMsg := "you must specify a name for the function or a Fully Qualified Function Name (FQFN)"
+	assert.NotNil(t, execErrMsg)
+	assert.True(t, strings.Contains(execErrMsg.Error(), failMsg))
+}

pkg/ctl/functions/functions.go

@@ -54,6 +54,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, querystateFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, putstateFunctionsCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, triggerFunctionsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, downloadFunctionsCmd)
 
 	return resourceCmd
 }

pkg/pulsar/functions.go

@@ -57,6 +57,24 @@ type Functions interface {
 	// DeleteFunction delete an existing function
 	DeleteFunction(tenant, namespace, name string) error
 
+	// Download Function Code
+	// @param destinationFile
+	//        file where data should be downloaded to
+	// @param path
+	//        Path where data is located
+	DownloadFunction(path, destinationFile string) error
+
+	// Download Function Code
+	// @param destinationFile
+	//        file where data should be downloaded to
+	// @param tenant
+	//        Tenant name
+	// @param namespace
+	//        Namespace name
+	// @param function
+	//        Function name
+	DownloadFunctionByNs(destinationFile, tenant, namespace, function string) error
+
 	// StartFunction start all function instances
 	StartFunction(tenant, namespace, name string) error
 
@@ -260,6 +278,46 @@ func (f *functions) DeleteFunction(tenant, namespace, name string) error {
 	return f.pulsar.Client.Delete(endpoint)
 }
 
+func (f *functions) DownloadFunction(path, destinationFile string) error {
+	endpoint := f.pulsar.endpoint(f.basePath, "download")
+	_, err := os.Open(destinationFile)
+	if err != nil {
+		_, err = os.Create(destinationFile)
+		if err != nil {
+			return err
+		}
+	}
+
+	tmpMap := make(map[string]string)
+	tmpMap["path"] = path
+
+	_, err = f.pulsar.Client.GetWithQueryParams(endpoint, nil, tmpMap, false)
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (f *functions) DownloadFunctionByNs(destinationFile, tenant, namespace, function string) error {
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, function, "download")
+	_, err := os.Open(destinationFile)
+	if err != nil {
+		_, err = os.Create(destinationFile)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = f.pulsar.Client.Get(endpoint, nil)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (f *functions) StartFunction(tenant, namespace, name string) error {
 	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
 	return f.pulsar.Client.Post(endpoint+"/start", "")

pkg/pulsar/utils/data.go

@@ -73,6 +73,8 @@ type FunctionData struct {
 	FunctionConfigFile   string          `json:"fnConfigFile"`
 	ProcessingGuarantees string          `json:"processingGuarantees"`
 	UserConfig           string          `json:"userConfigString"`
+	DestinationFile      string          `json:"destinationFile"`
+	Path                 string          `json:"path"`
 	FuncConf             *FunctionConfig `json:"-"`
 }